use crate::{
dex_connector::{slippage_price, string_to_decimal, DexConnector},
dex_request::{DexError, DexRequest, HttpMethod},
dex_websocket::DexWebSocket,
BalanceResponse, CreateOrderResponse, FilledOrder, FilledOrdersResponse, OrderSide,
TickerResponse,
};
use ::serde::{Deserialize, Serialize};
use async_trait::async_trait;
use debot_utils::parse_to_decimal;
use ethers::{signers::LocalWallet, types::H160};
use futures::{
stream::{SplitSink, SplitStream},
SinkExt, StreamExt,
};
use hyperliquid_rust_sdk_fork::{
BaseUrl, ClientCancelRequest, ClientLimit, ClientOrder, ClientOrderRequest, ExchangeClient,
ExchangeDataStatus, ExchangeResponseStatus,
};
use rust_decimal::prelude::*;
use rust_decimal::Decimal;
use std::{
collections::HashMap,
str::FromStr,
sync::{
atomic::{AtomicBool, Ordering},
Arc,
},
time::Duration,
};
use tokio::signal::unix::SignalKind;
use tokio::sync::Mutex;
use tokio::sync::RwLock;
use tokio::time::sleep;
use tokio::{net::TcpStream, task::JoinHandle};
use tokio::{select, signal::unix::signal};
use tokio_tungstenite::tungstenite::protocol::Message;
use tokio_tungstenite::MaybeTlsStream;
use tokio_tungstenite::WebSocketStream;
struct Config {
evm_wallet_address: String,
symbol_list: Vec<String>,
}
#[derive(Deserialize, Debug)]
struct SpotMetaToken {
#[serde(rename = "name")]
_name: String,
#[serde(rename = "szDecimals")]
_sz_decimals: u32,
#[serde(rename = "weiDecimals")]
_wei_decimals: u32,
#[serde(rename = "index")]
_index: usize,
}
#[derive(Deserialize, Debug, Clone)]
struct SpotMetaUniverse {
#[serde(rename = "name")]
name: String,
#[serde(rename = "tokens")]
_tokens: Vec<usize>,
#[serde(rename = "index")]
index: usize,
}
#[derive(Deserialize, Debug)]
struct SpotMetaResponse {
#[serde(rename = "tokens")]
_tokens: Vec<SpotMetaToken>,
#[serde(rename = "universe")]
universe: Vec<SpotMetaUniverse>,
}
#[derive(Serialize, Debug)]
struct InfoRequest<'a> {
#[serde(rename = "type")]
req_type: &'a str,
#[serde(skip_serializing_if = "Option::is_none")]
user: Option<&'a str>,
}
#[derive(Debug)]
struct TradeResult {
pub filled_side: OrderSide,
pub filled_size: Decimal,
pub filled_value: Decimal,
pub filled_fee: Decimal,
order_id: String,
}
#[derive(Default)]
struct DynamicMarketInfo {
pub market_price: Option<Decimal>,
pub min_tick: Option<Decimal>,
pub volume: Option<Decimal>,
pub num_trades: Option<u64>,
pub open_interest: Option<Decimal>,
pub funding_rate: Option<Decimal>,
pub oracle_price: Option<Decimal>,
}
struct StaticMarketInfo {
pub decimals: u32,
pub _max_leverage: u32,
}
pub struct HyperliquidConnector {
config: Config,
request: DexRequest,
web_socket: DexWebSocket,
running: Arc<AtomicBool>,
read_socket: Arc<Mutex<Option<SplitStream<WebSocketStream<MaybeTlsStream<TcpStream>>>>>>,
write_socket:
Arc<Mutex<Option<SplitSink<WebSocketStream<MaybeTlsStream<TcpStream>>, Message>>>>,
task_handle_read_message: Arc<Mutex<Option<JoinHandle<()>>>>,
task_handle_read_sigterm: Arc<Mutex<Option<JoinHandle<()>>>>,
trade_results: Arc<RwLock<HashMap<String, HashMap<String, TradeResult>>>>,
dynamic_market_info: Arc<RwLock<HashMap<String, DynamicMarketInfo>>>,
static_market_info: HashMap<String, StaticMarketInfo>,
spot_index_map: HashMap<String, usize>,
spot_reverse_map: Arc<HashMap<usize, String>>,
exchange_client: ExchangeClient,
}
#[derive(Debug)]
struct WebSocketMessage {
_channel: String,
data: WebSocketData,
}
#[derive(Debug)]
enum WebSocketData {
AllMidsData(AllMidsData),
UserFillsData(UserFillsData),
CandleData(CandleData),
ActiveAssetCtxData(ActiveAssetCtxData),
}
#[derive(Deserialize, Debug)]
struct AllMidsData {
mids: HashMap<String, String>,
}
#[allow(dead_code, non_snake_case)]
#[derive(Deserialize, Debug)]
struct CandleData {
t: u64, T: u64, s: String, i: String, o: Decimal, c: Decimal, h: Decimal, l: Decimal, v: Decimal, n: u64, }
#[derive(Deserialize, Debug)]
pub struct ActiveAssetCtxData {
pub coin: String, pub ctx: PerpsAssetCtx, }
#[allow(dead_code, non_snake_case)]
#[derive(Deserialize, Debug)]
pub struct PerpsAssetCtx {
pub dayNtlVlm: Decimal, pub prevDayPx: Decimal, pub markPx: Decimal, pub midPx: Option<Decimal>, pub funding: Decimal, pub openInterest: Decimal, pub oraclePx: Decimal, }
#[derive(Serialize, Deserialize, Debug)]
pub struct UserFillsData {
pub user: String,
pub fills: Vec<Fill>,
}
#[derive(Serialize, Deserialize, Debug)]
pub struct Fill {
pub coin: String,
pub px: Decimal,
pub sz: Decimal,
pub side: String,
pub dir: String,
#[serde(rename = "closedPnl")]
pub closed_pnl: Decimal,
pub oid: u64,
pub tid: u64,
pub fee: Decimal,
}
impl<'de> Deserialize<'de> for WebSocketMessage {
fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
where
D: serde::Deserializer<'de>,
{
#[derive(Deserialize)]
struct Helper {
channel: String,
data: serde_json::Value,
}
let helper = Helper::deserialize(deserializer)?;
let data = match helper.channel.as_str() {
"allMids" => {
let mids_data = AllMidsData::deserialize(helper.data)
.map(WebSocketData::AllMidsData)
.map_err(serde::de::Error::custom)?;
mids_data
}
"userFills" => {
let fills_data = UserFillsData::deserialize(helper.data)
.map(WebSocketData::UserFillsData)
.map_err(serde::de::Error::custom)?;
fills_data
}
"candle" => {
let candle_data = CandleData::deserialize(helper.data)
.map(WebSocketData::CandleData)
.map_err(serde::de::Error::custom)?;
candle_data
}
"activeAssetCtx" => {
let active_asset_ctx_data = ActiveAssetCtxData::deserialize(helper.data)
.map(WebSocketData::ActiveAssetCtxData)
.map_err(serde::de::Error::custom)?;
active_asset_ctx_data
}
_ => return Err(serde::de::Error::custom("unknown channel type")),
};
Ok(WebSocketMessage {
_channel: helper.channel,
data,
})
}
}
impl HyperliquidConnector {
pub async fn new(
rest_endpoint: &str,
web_socket_endpoint: &str,
private_key: &str,
evm_wallet_address: &str,
vault_address: Option<String>,
use_agent: bool,
agent_name: Option<String>,
symbol_list: &[&str],
) -> Result<Self, DexError> {
let request = DexRequest::new(rest_endpoint.to_owned()).await?;
let web_socket = DexWebSocket::new(web_socket_endpoint.to_owned());
let evm_wallet_address = vault_address
.clone()
.unwrap_or_else(|| evm_wallet_address.into());
let config = Config {
evm_wallet_address,
symbol_list: symbol_list.iter().map(|s| s.to_string()).collect(),
};
let vault_address: Option<H160> = vault_address
.as_deref()
.and_then(|v| H160::from_str(v).ok());
let mut local_wallet: LocalWallet = private_key.parse().unwrap();
if use_agent {
let ec_tmp =
ExchangeClient::new(None, local_wallet, Some(BaseUrl::Mainnet), None, None)
.await
.map_err(|e| DexError::Other(e.to_string()))?;
let (pk, resp) = ec_tmp
.approve_agent(None, agent_name)
.await
.map_err(|e| DexError::Other(e.to_string()))?;
log::info!("Agent approved: {resp:?}");
local_wallet = pk.parse().unwrap();
}
let exchange_client = ExchangeClient::new(
None,
local_wallet,
Some(BaseUrl::Mainnet),
None,
vault_address,
)
.await
.map_err(|e| DexError::Other(e.to_string()))?;
let mut instance = HyperliquidConnector {
config,
request,
web_socket,
trade_results: Arc::new(RwLock::new(HashMap::new())),
running: Arc::new(AtomicBool::new(false)),
read_socket: Arc::new(Mutex::new(None)),
write_socket: Arc::new(Mutex::new(None)),
task_handle_read_message: Arc::new(Mutex::new(None)),
task_handle_read_sigterm: Arc::new(Mutex::new(None)),
dynamic_market_info: Arc::new(RwLock::new(HashMap::new())),
static_market_info: HashMap::new(),
spot_index_map: HashMap::new(),
spot_reverse_map: Arc::new(HashMap::new()),
exchange_client,
};
instance.retrive_market_metadata().await?;
let info_payload = serde_json::to_string(&InfoRequest {
req_type: "spotMeta",
user: None,
})
.map_err(|e| DexError::Other(e.to_string()))?;
let spot_meta: SpotMetaResponse = instance
.request
.handle_request::<SpotMetaResponse, InfoRequest<'_>>(
HttpMethod::Post,
"/info".into(),
&HashMap::new(),
info_payload,
)
.await?;
let token_name_map: HashMap<usize, String> = spot_meta
._tokens
.iter()
.map(|t| (t._index, t._name.clone()))
.collect();
let mut idx_from_pair = HashMap::<String, usize>::new();
let mut pair_from_idx = HashMap::<usize, String>::new();
for uni in &spot_meta.universe {
let pair = if !uni.name.starts_with('@') {
uni.name.clone()
} else if uni._tokens.len() == 2 {
format!(
"{}/{}",
token_name_map.get(&uni._tokens[0]).unwrap_or(&"?".into()),
token_name_map.get(&uni._tokens[1]).unwrap_or(&"?".into())
)
} else {
log::warn!(
"universe idx {} has unexpected token vec {:?}",
uni.index,
uni._tokens
);
uni.name.clone()
};
idx_from_pair.insert(pair.clone(), uni.index);
pair_from_idx.insert(uni.index, pair);
}
instance.spot_index_map = idx_from_pair;
instance.spot_reverse_map = Arc::new(pair_from_idx);
Ok(instance)
}
pub async fn start_web_socket(&self) -> Result<(), DexError> {
log::info!("start_web_socket");
let (write, read) = self
.web_socket
.clone()
.connect()
.await
.map_err(|_| DexError::Other("Failed to connect to WebSocket".to_string()))?;
{
let mut read_lock = self.read_socket.lock().await;
*read_lock = Some(read);
}
{
let mut write_lock = self.write_socket.lock().await;
*write_lock = Some(write);
}
self.running.store(true, Ordering::SeqCst);
self.subscribe_to_channels(&self.config.evm_wallet_address)
.await?;
let running = self.running.clone();
let read_sock = self.read_socket.clone();
let write_sock = self.write_socket.clone();
let dmi = self.dynamic_market_info.clone();
let trs = self.trade_results.clone();
let rev_map = self.spot_reverse_map.clone();
let reader_handle = tokio::spawn(async move {
let mut idle_counter = 0;
while running.load(Ordering::SeqCst) {
if let Some(stream) = read_sock.lock().await.as_mut() {
tokio::select! {
msg = stream.next() => match msg {
Some(Ok(Message::Text(txt))) => {
idle_counter = 0;
if txt == "{}" {
if let Some(w) = write_sock.lock().await.as_mut() {
let _ = w.send(Message::Text(txt)).await;
}
} else {
if let Err(e) = HyperliquidConnector::handle_websocket_message(
Message::Text(txt),
dmi.clone(),
trs.clone(),
rev_map.clone(),
).await {
log::error!("WebSocket handler error: {:?}", e);
break;
}
}
}
Some(Ok(_)) => {
}
Some(Err(err)) => {
log::error!("WebSocket read error: {:?}", err);
break;
}
None => {
log::info!("WebSocket stream closed");
break;
}
},
_ = tokio::time::sleep(Duration::from_secs(10)) => {
idle_counter += 1;
if idle_counter >= 10 {
log::error!("No WebSocket messages for 100s, shutting down reader");
break;
}
}
}
}
}
running.store(false, Ordering::SeqCst);
log::info!("WebSocket reader task ended");
});
*self.task_handle_read_message.lock().await = Some(reader_handle);
let running_for_sig = self.running.clone();
let sig_handle = tokio::spawn(async move {
let mut sigterm =
signal(SignalKind::terminate()).expect("Failed to bind SIGTERM handler");
loop {
select! {
_ = sigterm.recv() => {
log::info!("SIGTERM received, stopping WebSocket");
running_for_sig.store(false, Ordering::SeqCst);
break;
}
_ = tokio::time::sleep(Duration::from_secs(1)) => {
if !running_for_sig.load(Ordering::SeqCst) {
break;
}
}
}
}
});
*self.task_handle_read_sigterm.lock().await = Some(sig_handle);
Ok(())
}
pub async fn stop_web_socket(&self) -> Result<(), DexError> {
log::info!("stop_web_socket");
self.running.store(false, Ordering::SeqCst);
{
let mut write_guard = self.write_socket.lock().await;
if let Some(write_socket) = write_guard.as_mut() {
if let Err(e) = write_socket.send(Message::Close(None)).await {
log::error!("Failed to send WebSocket close message: {:?}", e);
}
}
*write_guard = None;
}
{
let mut read_guard = self.read_socket.lock().await;
*read_guard = None;
}
if let Some(handle) = self.task_handle_read_message.lock().await.take() {
let _ = handle.await;
}
if let Some(handle) = self.task_handle_read_sigterm.lock().await.take() {
let _ = handle.await;
}
drop(self.web_socket.clone());
Ok(())
}
async fn subscribe_to_channels(&self, user_address: &str) -> Result<(), DexError> {
let all_mids_subscription = serde_json::json!({
"method": "subscribe",
"subscription": {
"type": "allMids"
}
})
.to_string();
let user_fills_subscription = serde_json::json!({
"method": "subscribe",
"subscription": {
"type": "userFills",
"user": user_address
}
})
.to_string();
let mut write_socket_lock = self.write_socket.lock().await;
if let Some(write_socket) = write_socket_lock.as_mut() {
if let Err(e) = write_socket
.send(Message::Text(all_mids_subscription))
.await
{
return Err(DexError::WebSocketError(format!(
"Failed to subscribe to allMids: {}",
e
)));
}
if let Err(e) = write_socket
.send(Message::Text(user_fills_subscription))
.await
{
return Err(DexError::WebSocketError(format!(
"Failed to subscribe to userFills: {}",
e
)));
}
for symbol in &self.config.symbol_list {
let coin = resolve_coin(symbol, &self.spot_index_map);
let candle_subscription = serde_json::json!({
"method": "subscribe",
"subscription": {
"type": "candle",
"coin": coin,
"interval": "1m"
}
})
.to_string();
if let Err(e) = write_socket.send(Message::Text(candle_subscription)).await {
return Err(DexError::WebSocketError(format!(
"Failed to subscribe to candle for {}: {}",
symbol, e
)));
}
let active_asset_ctx_subscription = serde_json::json!({
"method": "subscribe",
"subscription": {
"type": "activeAssetCtx",
"coin": coin,
}
})
.to_string();
if let Err(e) = write_socket
.send(Message::Text(active_asset_ctx_subscription))
.await
{
return Err(DexError::WebSocketError(format!(
"Failed to subscribe to activeAssetCtx: {}",
e
)));
}
}
} else {
return Err(DexError::WebSocketError(
"Write socket is not available".to_string(),
));
}
Ok(())
}
async fn handle_websocket_message(
msg: Message,
dynamic_market_info: Arc<RwLock<HashMap<String, DynamicMarketInfo>>>,
trade_results: Arc<RwLock<HashMap<String, HashMap<String, TradeResult>>>>,
spot_reverse_map: Arc<HashMap<usize, String>>,
) -> Result<(), DexError> {
if let Message::Text(text) = msg {
for line in text.split('\n') {
if line.is_empty() {
continue;
}
if let Ok(message) = serde_json::from_str::<WebSocketMessage>(line) {
match message.data {
WebSocketData::AllMidsData(ref data) => {
Self::process_all_mids_message(
data,
dynamic_market_info.clone(),
spot_reverse_map.clone(),
)
.await;
}
WebSocketData::CandleData(ref data) => {
Self::process_candle_message(
data,
dynamic_market_info.clone(),
spot_reverse_map.clone(),
)
.await;
}
WebSocketData::UserFillsData(ref data) => {
Self::process_account_data(data, trade_results.clone()).await;
}
WebSocketData::ActiveAssetCtxData(ref data) => {
Self::process_active_asset_ctx_message(
data,
dynamic_market_info.clone(),
spot_reverse_map.clone(),
)
.await;
}
}
}
}
}
Ok(())
}
async fn process_all_mids_message(
mids_data: &AllMidsData,
dynamic_market_info: Arc<RwLock<HashMap<String, DynamicMarketInfo>>>,
spot_reverse_map: Arc<HashMap<usize, String>>,
) {
for (raw_coin, mid_price_str) in &mids_data.mids {
let coin = if let Some(stripped) = raw_coin.strip_prefix('@') {
stripped
.parse::<usize>()
.ok()
.and_then(|idx| spot_reverse_map.get(&idx).cloned())
.unwrap_or_else(|| {
log::warn!("spot_reverse_map に {} が無い (@{})", raw_coin, stripped);
raw_coin.clone()
})
} else {
raw_coin.clone()
};
let market_key = if coin.contains('/') || coin.contains('-') {
coin.clone() } else {
format!("{}-USD", coin) };
if let Ok(mid) = string_to_decimal(Some(mid_price_str.clone())) {
let mut guard = dynamic_market_info.write().await;
let info = guard.entry(market_key.clone()).or_default();
if info.min_tick.is_none() {
info.min_tick = Some(Self::calculate_min_tick(mid));
}
info.market_price = Some(mid);
if market_key == "UBTC/USDC" {
log::info!("mid update UBTC/USDC → {}", mid);
}
}
}
}
async fn process_candle_message(
candle: &CandleData,
dynamic_market_info: Arc<RwLock<HashMap<String, DynamicMarketInfo>>>,
spot_reverse_map: Arc<HashMap<usize, String>>,
) {
let coin = if let Some(stripped) = candle.s.strip_prefix('@') {
stripped
.parse::<usize>()
.ok()
.and_then(|idx| spot_reverse_map.get(&idx).cloned())
.unwrap_or_else(|| {
log::warn!(
"in spot_reverse_map: {} is missing (@{})",
candle.s,
stripped
);
candle.s.clone()
})
} else {
candle.s.clone()
};
let market_key = if coin.contains('/') || coin.contains('-') {
coin.clone()
} else {
format!("{}-USD", coin)
};
let mut guard = dynamic_market_info.write().await;
let info = guard.entry(market_key.clone()).or_default();
info.volume = Some(candle.v);
info.num_trades = Some(candle.n);
}
async fn process_active_asset_ctx_message(
asset_data: &ActiveAssetCtxData,
dynamic_market_info: Arc<RwLock<HashMap<String, DynamicMarketInfo>>>,
spot_reverse_map: Arc<HashMap<usize, String>>,
) {
let coin = if let Some(stripped) = asset_data.coin.strip_prefix('@') {
stripped
.parse::<usize>()
.ok()
.and_then(|idx| spot_reverse_map.get(&idx).cloned())
.unwrap_or_else(|| {
log::warn!(
"in spot_reverse_map {} is missing (@{})",
asset_data.coin,
stripped
);
asset_data.coin.clone()
})
} else {
asset_data.coin.clone()
};
let market_key = if coin.contains('/') || coin.contains('-') {
coin.clone()
} else {
format!("{}-USD", coin)
};
let mut guard = dynamic_market_info.write().await;
let info = guard
.entry(market_key.clone())
.or_insert_with(DynamicMarketInfo::default);
info.funding_rate = Some(asset_data.ctx.funding);
info.open_interest = Some(asset_data.ctx.openInterest);
info.oracle_price = Some(asset_data.ctx.oraclePx);
}
async fn process_account_data(
data: &UserFillsData,
trade_results: Arc<RwLock<HashMap<String, HashMap<String, TradeResult>>>>,
) {
for fill in &data.fills {
log::debug!("{:?}", fill);
let filled_side = if fill.side == "A" {
OrderSide::Short
} else {
OrderSide::Long
};
let filled_size = fill.sz;
let filled_price = fill.px;
let filled_value = filled_size * filled_price;
let filled_fee = fill.fee;
let order_id = fill.oid;
let trade_id = fill.tid;
let market_id = if fill.coin.contains('/') || fill.coin.contains('-') {
fill.coin.clone()
} else {
format!("{}-USD", fill.coin)
};
let trade_result = TradeResult {
filled_side,
filled_size,
filled_value,
filled_fee,
order_id: order_id.to_string(),
};
let mut trade_results_guard = trade_results.write().await;
trade_results_guard
.entry(market_id.clone())
.or_default()
.insert(trade_id.to_string(), trade_result);
}
}
}
#[derive(Serialize, Debug, Clone)]
struct HyperliquidDefaultPayload {
r#type: String,
#[serde(skip_serializing_if = "Option::is_none")]
user: Option<String>,
}
#[derive(Deserialize, Debug)]
struct HyperliquidRetrieveUserStateResponse {
#[serde(rename = "marginSummary")]
margin_summary: Option<HyperliquidMarginSummary>,
}
#[derive(Deserialize, Debug)]
struct HyperliquidMarginSummary {
#[serde(rename = "accountValue")]
account_value: String,
#[serde(rename = "totalRawUsd")]
total_rawusd: String,
}
#[derive(Deserialize, Debug)]
struct HyperliquidRetriveUserOpenOrder {
coin: String,
oid: u64,
}
#[derive(Deserialize, Debug)]
struct HyperliquidRetriveUserPositionResponse {
#[serde(rename = "assetPositions")]
asset_positions: Vec<HyperliquidRetriveUserPositionResponseBody>,
}
#[derive(Deserialize, Debug)]
struct HyperliquidRetriveUserPositionResponseBody {
position: HyperliquidRetriveUserPosition,
}
#[derive(Deserialize, Debug)]
struct HyperliquidRetriveUserPosition {
coin: String,
szi: Decimal,
}
#[derive(Deserialize, Debug)]
struct HyperliquidRetriveMarketMetadataResponse {
universe: Vec<HyperliquidRetriveMarketMetadata>,
}
#[derive(Deserialize, Debug)]
struct HyperliquidRetriveMarketMetadata {
name: String,
#[serde(rename = "szDecimals")]
decimals: u32,
#[serde(rename = "maxLeverage")]
max_leverage: u32,
}
#[async_trait]
impl DexConnector for HyperliquidConnector {
async fn start(&self) -> Result<(), DexError> {
self.start_web_socket().await?;
sleep(Duration::from_secs(5)).await;
Ok(())
}
async fn stop(&self) -> Result<(), DexError> {
self.stop_web_socket().await?;
Ok(())
}
async fn restart(&self, max_retries: i32) -> Result<(), DexError> {
log::info!("Restarting WebSocket connection...");
let mut retry_count = 0;
let mut backoff_delay = Duration::from_secs(1);
while retry_count < max_retries {
if let Err(e) = self.stop_web_socket().await {
log::error!(
"Failed to stop WebSocket on attempt {}: {:?}",
retry_count + 1,
e
);
} else {
log::info!(
"Successfully stopped WebSocket on attempt {}.",
retry_count + 1
);
}
sleep(backoff_delay).await;
match self.start_web_socket().await {
Ok(_) => {
log::info!(
"Successfully started WebSocket on attempt {}.",
retry_count + 1
);
return Ok(());
}
Err(e) => {
log::error!(
"Failed to start WebSocket on attempt {}: {:?}",
retry_count + 1,
e
);
retry_count += 1;
backoff_delay *= 2; }
}
}
log::error!(
"Failed to restart WebSocket after {} attempts.",
max_retries
);
Err(DexError::Other(format!(
"Failed to restart WebSocket after {} attempts.",
max_retries
)))
}
async fn set_leverage(&self, symbol: &str, leverage: u32) -> Result<(), DexError> {
let asset = Self::extract_asset_name(symbol);
self.exchange_client
.update_leverage(leverage, asset, false, None)
.await
.map_err(|e| DexError::Other(e.to_string()))?;
Ok(())
}
async fn get_ticker(
&self,
symbol: &str,
_test_price: Option<Decimal>,
) -> Result<TickerResponse, DexError> {
if !self.running.load(Ordering::SeqCst) {
return Err(DexError::NoConnection);
}
let dynamic_info_guard = self.dynamic_market_info.read().await;
let dynamic_info = dynamic_info_guard
.get(symbol)
.ok_or_else(|| DexError::Other("No dynamic market info available".to_string()))?;
let price = dynamic_info
.market_price
.ok_or_else(|| DexError::Other("No price available".to_string()))?;
let min_tick = dynamic_info.min_tick;
let volume = dynamic_info.volume;
let num_trades = dynamic_info.num_trades;
let funding_rate = dynamic_info.funding_rate;
let open_interest = dynamic_info.open_interest;
let oracle_price = dynamic_info.oracle_price;
Ok(TickerResponse {
symbol: symbol.to_owned(),
price,
min_tick,
min_order: None,
volume,
num_trades,
funding_rate,
open_interest,
oracle_price,
})
}
async fn get_filled_orders(&self, symbol: &str) -> Result<FilledOrdersResponse, DexError> {
let mut response: Vec<FilledOrder> = vec![];
let trade_results_guard = self.trade_results.read().await;
let orders = match trade_results_guard.get(symbol) {
Some(v) => v,
None => return Ok(FilledOrdersResponse::default()),
};
for (trade_id, order) in orders.iter() {
let filled_order = FilledOrder {
order_id: order.order_id.clone(),
trade_id: trade_id.clone(),
is_rejected: false,
filled_side: Some(order.filled_side.clone()),
filled_size: Some(order.filled_size),
filled_fee: Some(order.filled_fee),
filled_value: Some(order.filled_value),
};
response.push(filled_order);
}
Ok(FilledOrdersResponse { orders: response })
}
async fn get_balance(&self) -> Result<BalanceResponse, DexError> {
let request_url = "/info";
let action = HyperliquidDefaultPayload {
r#type: "clearinghouseState".to_owned(),
user: Some(self.config.evm_wallet_address.clone()),
};
let res = self
.handle_request_with_action::<HyperliquidRetrieveUserStateResponse, HyperliquidDefaultPayload>(
request_url.to_string(),
&action,
)
.await?;
if let Some(summary) = res.margin_summary {
let equity = match parse_to_decimal(&summary.account_value) {
Ok(v) => v,
Err(e) => return Err(DexError::Other(format!("acount_equity: {:?}", e))),
};
let balance = match parse_to_decimal(&summary.total_rawusd) {
Ok(v) => v,
Err(e) => return Err(DexError::Other(format!("balance: {:?}", e))),
};
Ok(BalanceResponse {
equity: equity,
balance: balance,
})
} else {
return Err(DexError::Other(String::from("Unknown error")));
}
}
async fn clear_filled_order(&self, symbol: &str, order_id: &str) -> Result<(), DexError> {
let mut trade_results_guard = self.trade_results.write().await;
if let Some(orders) = trade_results_guard.get_mut(symbol) {
if orders.contains_key(order_id) {
orders.remove(order_id);
} else {
return Err(DexError::Other(format!(
"filled order(order_id:{}({})) does not exist",
order_id, symbol
)));
}
} else {
return Err(DexError::Other(format!(
"filled order(symbol:{}({})) does not exist",
symbol, order_id
)));
}
Ok(())
}
async fn clear_all_filled_order(&self) -> Result<(), DexError> {
let mut trade_results_guard = self.trade_results.write().await;
trade_results_guard.clear();
Ok(())
}
async fn create_order(
&self,
symbol: &str,
size: Decimal,
side: OrderSide,
price: Option<Decimal>,
spread: Option<i64>,
) -> Result<CreateOrderResponse, DexError> {
let (price, time_in_force) = match price {
Some(v) => (v, "Alo"),
None => {
let price = self.get_worst_price(symbol, &side).await?;
(price, "Ioc")
}
};
let dynamic_market_info_guard = self.dynamic_market_info.read().await;
let market_info = dynamic_market_info_guard
.get(symbol)
.ok_or_else(|| DexError::Other("Market info not found".to_string()))?;
let min_tick = market_info
.min_tick
.ok_or_else(|| DexError::Other("Min tick not set for market".to_string()))?;
let rounded_price = Self::round_price(price, min_tick, side.clone(), spread);
let rounded_size = self.floor_size(size, symbol);
log::debug!("{}, {}({}), {}", symbol, rounded_price, price, rounded_size,);
let asset = resolve_coin(symbol, &self.spot_index_map);
let order = ClientOrderRequest {
asset,
is_buy: side == OrderSide::Long,
reduce_only: false,
limit_px: rounded_price
.to_f64()
.ok_or_else(|| DexError::Other("Conversion to f64 failed".to_string()))?,
sz: rounded_size
.to_f64()
.ok_or_else(|| DexError::Other("Conversion to f64 failed".to_string()))?,
cloid: None,
order_type: ClientOrder::Limit(ClientLimit {
tif: time_in_force.to_string(),
}),
};
let res = self
.exchange_client
.order(order, None)
.await
.map_err(|e| DexError::Other(e.to_string()))?;
let res = match res {
ExchangeResponseStatus::Ok(exchange_response) => exchange_response,
ExchangeResponseStatus::Err(e) => return Err(DexError::ServerResponse(e.to_string())),
};
let status = res.data.unwrap().statuses[0].clone();
let order_id = match status {
ExchangeDataStatus::Filled(order) => order.oid,
ExchangeDataStatus::Resting(order) => order.oid,
_ => {
return Err(DexError::ServerResponse(
"Unknown ExchangeDataStaus".to_owned(),
))
}
};
Ok(CreateOrderResponse {
order_id: order_id.to_string(),
ordered_price: rounded_price,
ordered_size: rounded_size,
})
}
async fn cancel_order(&self, symbol: &str, order_id: &str) -> Result<(), DexError> {
let asset = resolve_coin(symbol, &self.spot_index_map);
let cancel = ClientCancelRequest {
asset,
oid: u64::from_str(order_id).unwrap_or_default(),
};
self.exchange_client
.cancel(cancel, None)
.await
.map_err(|e| DexError::Other(e.to_string()))?;
Ok(())
}
async fn cancel_all_orders(&self, symbol: Option<String>) -> Result<(), DexError> {
let open_orders = self.get_orders().await?;
let order_ids: Vec<String> = open_orders
.iter()
.filter(|order| {
symbol.as_deref() == Some(&format!("{}-USD", order.coin)) || symbol.is_none()
})
.map(|order| order.oid.to_string())
.collect();
self.cancel_orders(symbol, order_ids).await
}
async fn cancel_orders(
&self,
symbol: Option<String>,
order_ids: Vec<String>,
) -> Result<(), DexError> {
let open_orders = self.get_orders().await?;
let mut cancels = Vec::new();
for order in open_orders {
let order_symbol = format!("{}-USD", order.coin);
if (symbol.as_deref() == Some(&order_symbol) || symbol.is_none())
&& order_ids.contains(&order.oid.to_string())
{
cancels.push(ClientCancelRequest {
asset: Self::extract_asset_name(&order_symbol).to_owned(),
oid: order.oid,
});
}
}
if !cancels.is_empty() {
if let Err(e) = self.exchange_client.bulk_cancel(cancels, None).await {
log::error!("cancel_orders: Failed to cancel orders: {:?}", e);
}
}
Ok(())
}
async fn close_all_positions(&self, symbol: Option<String>) -> Result<(), DexError> {
let open_positions = self.get_positions().await?;
log::warn!("close_all_positions: symbol = {:?}", symbol);
for p in open_positions {
let position = p.position;
let order_symbol = format!("{}-USD", position.coin);
if symbol.as_deref() == Some(&order_symbol) || symbol.is_none() {
let reversed_side = if position.szi.is_sign_negative() {
OrderSide::Long
} else {
OrderSide::Short
};
let size = position.szi.abs();
if let Err(e) = self
.create_order(&order_symbol, size, reversed_side, None, None)
.await
{
log::error!("close_all_positions: {:?}", e);
}
}
}
Ok(())
}
async fn clear_last_trades(&self, _symbol: &str) -> Result<(), DexError> {
Ok(())
}
}
impl HyperliquidConnector {
async fn handle_request_with_action<T, U>(
&self,
request_url: String,
action: &U,
) -> Result<T, DexError>
where
T: for<'de> Deserialize<'de>,
U: Serialize + std::fmt::Debug + Clone,
{
let json_payload =
serde_json::to_value(action).map_err(|e| DexError::Other(e.to_string()))?;
log::debug!("json_payload = {:?}", json_payload);
self.request
.handle_request::<T, U>(
HttpMethod::Post,
request_url,
&HashMap::new(),
json_payload.to_string(),
)
.await
.map_err(|e| DexError::Other(e.to_string()))
}
async fn get_positions(
&self,
) -> Result<Vec<HyperliquidRetriveUserPositionResponseBody>, DexError> {
let request_url = "/info";
let action = HyperliquidDefaultPayload {
r#type: "clearinghouseState".to_owned(),
user: Some(self.config.evm_wallet_address.clone()),
};
let res: HyperliquidRetriveUserPositionResponse = self
.handle_request_with_action::<HyperliquidRetriveUserPositionResponse, HyperliquidDefaultPayload>(
request_url.to_string(),
&action,
)
.await?;
Ok(res.asset_positions)
}
async fn get_orders(&self) -> Result<Vec<HyperliquidRetriveUserOpenOrder>, DexError> {
let request_url = "/info";
let action = HyperliquidDefaultPayload {
r#type: "openOrders".to_owned(),
user: Some(self.config.evm_wallet_address.clone()),
};
let res: Vec<HyperliquidRetriveUserOpenOrder> = self
.handle_request_with_action::<Vec<HyperliquidRetriveUserOpenOrder>, HyperliquidDefaultPayload>(
request_url.to_string(),
&action,
)
.await?;
Ok(res)
}
async fn retrive_market_metadata(&mut self) -> Result<(), DexError> {
let request_url = "/info";
let action = HyperliquidDefaultPayload {
r#type: "meta".to_owned(),
user: None,
};
let res = self
.handle_request_with_action::<HyperliquidRetriveMarketMetadataResponse, HyperliquidDefaultPayload>(
request_url.to_string(),
&action,
)
.await?;
let mut static_market_info_update = HashMap::new();
for metadata in res.universe.into_iter() {
let market_id = format!("{}-USD", metadata.name);
static_market_info_update.insert(
market_id,
StaticMarketInfo {
decimals: metadata.decimals,
_max_leverage: metadata.max_leverage,
},
);
}
self.static_market_info = static_market_info_update;
Ok(())
}
async fn get_worst_price(&self, symbol: &str, side: &OrderSide) -> Result<Decimal, DexError> {
let market_price = self.get_market_price(symbol).await?;
let worst_price = slippage_price(market_price, *side == OrderSide::Long);
Ok(worst_price)
}
async fn get_market_price(&self, symbol: &str) -> Result<Decimal, DexError> {
let market_info_guard = self.dynamic_market_info.read().await;
match market_info_guard.get(symbol) {
Some(v) => match v.market_price {
Some(price) => Ok(price),
None => Err(DexError::Other("Price is None".to_string())),
},
None => Err(DexError::Other("No price available".to_string())),
}
}
fn calculate_min_tick(price: Decimal) -> Decimal {
let price_str = price.to_string();
let parts: Vec<&str> = price_str.split('.').collect();
let integer_part = parts[0];
if integer_part.len() >= 5 {
return Decimal::ONE;
}
let scale = 5 - integer_part.len();
Decimal::new(1, scale as u32)
}
fn round_price(
price: Decimal,
min_tick: Decimal,
order_side: OrderSide,
spread: Option<i64>,
) -> Decimal {
if min_tick.is_zero() {
log::error!("round_price: min_tick is zero");
return price;
}
let spread = match spread {
Some(v) => Decimal::new(v, 0),
None => Decimal::ZERO,
};
match order_side {
OrderSide::Long => (price / min_tick - spread).floor() * min_tick,
OrderSide::Short => (price / min_tick + spread).ceil() * min_tick,
}
}
fn floor_size(&self, size: Decimal, symbol: &str) -> Decimal {
let decimals = match self.static_market_info.get(symbol) {
Some(v) => v.decimals,
None => {
log::error!("symbol meta is not available: {}", symbol);
return size;
}
};
size.round_dp(decimals)
}
fn extract_asset_name(symbol: &str) -> &str {
symbol.split('-').next().unwrap_or(symbol)
}
}
fn resolve_coin(sym: &str, map: &HashMap<String, usize>) -> String {
if sym.contains('/') {
match map.get(sym) {
Some(idx) => format!("@{}", idx),
None => {
log::warn!("resolve_coin: {} is not in spot_index_map", sym);
sym.to_string()
}
}
} else {
sym.to_string()
}
}