From 5d4869e8559ef11455c33507d0a0397e10d9a676 Mon Sep 17 00:00:00 2001 From: Matthew O'Connell Date: Thu, 14 May 2026 17:34:56 +0200 Subject: [PATCH 1/8] add implementation --- Cargo.lock | 88 +++++++++ Cargo.toml | 3 + src/.env.example | 14 ++ src/exchanges/hyperliquid/balances.rs | 30 ++++ src/exchanges/hyperliquid/client.rs | 26 +-- src/exchanges/hyperliquid/market_data.rs | 19 ++ src/exchanges/hyperliquid/mod.rs | 2 + src/exchanges/hyperliquid/orders.rs | 120 +++++++++++++ src/execution.rs | 96 +++++++++- src/lib.rs | 1 + src/main.rs | 27 ++- src/market_data.rs | 11 +- src/order_state.rs | 219 ++++++++++++++++++++++- src/risk.rs | 1 + src/weights.rs | 33 ++++ 15 files changed, 652 insertions(+), 38 deletions(-) create mode 100644 src/.env.example create mode 100644 src/exchanges/hyperliquid/orders.rs create mode 100644 src/weights.rs diff --git a/Cargo.lock b/Cargo.lock index f69cd08..378ed14 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1539,8 +1539,11 @@ dependencies = [ "async-trait", "ethers", "hyperliquid_rust_sdk", + "rand 0.8.6", "serde", "tokio", + "tracing", + "tracing-subscriber", ] [[package]] @@ -1965,6 +1968,15 @@ version = "0.4.29" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5e5032e24019045c762d3c0f28f5b6b8bbf38563a65908389bf7978758920897" +[[package]] +name = "matchers" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d1525a2a28c7f4fa0fc98bb91ae755d1e2d1505079e05539e35bc876b5d65ae9" +dependencies = [ + "regex-automata", +] + [[package]] name = "md-5" version = "0.10.6" @@ -2031,6 +2043,15 @@ version = "1.0.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "650eef8c711430f1a879fdd01d4745a7deea475becfb90269c06775983bbf086" +[[package]] +name = "nu-ansi-term" +version = "0.50.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5" +dependencies = [ + "windows-sys 0.61.2", +] + [[package]] name = "num-bigint" version = "0.4.6" @@ -3075,6 +3096,15 @@ dependencies = [ "keccak", ] +[[package]] +name = "sharded-slab" +version = "0.1.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f40ca3c46823713e0d4209592e8d6e826aa57e928f09752619fc696c499637f6" +dependencies = [ + "lazy_static", +] + [[package]] name = "shlex" version = "1.3.0" @@ -3398,6 +3428,15 @@ dependencies = [ "syn 2.0.117", ] +[[package]] +name = "thread_local" +version = "1.1.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f60246a4944f24f6e018aa17cdeffb7818b76356965d03b07d6a9886e8962185" +dependencies = [ + "cfg-if", +] + [[package]] name = "time" version = "0.3.47" @@ -3647,6 +3686,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "db97caf9d906fbde555dd62fa95ddba9eecfd14cb388e4f491a66d74cd5fb79a" dependencies = [ "once_cell", + "valuable", ] [[package]] @@ -3659,6 +3699,48 @@ dependencies = [ "tracing", ] +[[package]] +name = "tracing-log" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ee855f1f400bd0e5c02d150ae5de3840039a3f54b025156404e34c23c03f47c3" +dependencies = [ + "log", + "once_cell", + "tracing-core", +] + +[[package]] +name = "tracing-serde" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "704b1aeb7be0d0a84fc9828cae51dab5970fee5088f83d1dd7ee6f6246fc6ff1" +dependencies = [ + "serde", + "tracing-core", +] + +[[package]] +name = "tracing-subscriber" +version = "0.3.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cb7f578e5945fb242538965c2d0b04418d38ec25c79d160cd279bf0731c8d319" +dependencies = [ + "matchers", + "nu-ansi-term", + "once_cell", + "regex-automata", + "serde", + "serde_json", + "sharded-slab", + "smallvec", + "thread_local", + "tracing", + "tracing-core", + "tracing-log", + "tracing-serde", +] + [[package]] name = "try-lock" version = "0.2.5" @@ -3779,6 +3861,12 @@ dependencies = [ "wasm-bindgen", ] +[[package]] +name = "valuable" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ba73ea9cf16a25df0c8caa16c51acb937d5712a8429db78a3ee29d5dcacd3a65" + [[package]] name = "vcpkg" version = "0.2.15" diff --git a/Cargo.toml b/Cargo.toml index 3e681a9..c338321 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -8,5 +8,8 @@ anyhow = "1.0.102" async-trait = "0.1.89" ethers = "2.0.14" hyperliquid_rust_sdk = "0.6.0" +rand = "0.8.5" serde = "1.0.228" tokio = { version = "1.52.1", features = ["full"] } +tracing = "0.1.41" +tracing-subscriber = { version = "0.3.20", features = ["env-filter", "fmt", "json"] } diff --git a/src/.env.example b/src/.env.example new file mode 100644 index 0000000..19fe38e --- /dev/null +++ b/src/.env.example @@ -0,0 +1,14 @@ +HL_BASE_URL=https://api.hyperliquid.xyz +WALLET_ADDRESS=0000000000000000000000000000000000000000 +PRIVATE_KEY= + +USE_MAINNET=false +ALLOW_ORDER=false + +SYMBOLS=ETH,BTC +QUOTE_ASSET=USDC + +MAX_ORDER_USD=25 +MAX_POSITION_USD=100 + +LOG_LEVEL=info \ No newline at end of file diff --git a/src/exchanges/hyperliquid/balances.rs b/src/exchanges/hyperliquid/balances.rs index e69de29..c21f730 100644 --- a/src/exchanges/hyperliquid/balances.rs +++ b/src/exchanges/hyperliquid/balances.rs @@ -0,0 +1,30 @@ +use ethers::types::H160; +use hyperliquid_rust_sdk::InfoClient; + +use crate::wallet::WalletHolding; + +pub async fn get_wallet_holdings( + info_client: &InfoClient, + wallet_address: H160, +) -> anyhow::Result> { + let user_state = info_client.user_state(wallet_address).await?; + + Ok(user_state + .asset_positions + .into_iter() + .map(|asset_position| { + let position = asset_position.position; + + WalletHolding { + coin: position.coin, + total: position.szi, + entry_px: position.entry_px, + leverage: Some(position.leverage.value.to_string()), + unrealized_pnl: Some(position.unrealized_pnl), + realised_pnl: None, + funding_unlocked: Some(position.cum_funding.since_open), + collateral: Some(position.margin_used), + } + }) + .collect()) +} diff --git a/src/exchanges/hyperliquid/client.rs b/src/exchanges/hyperliquid/client.rs index f35f9ca..e720b14 100644 --- a/src/exchanges/hyperliquid/client.rs +++ b/src/exchanges/hyperliquid/client.rs @@ -12,7 +12,7 @@ use crate::{ wallet::WalletHolding, }; -use super::market_data; +use super::{balances, market_data, orders}; pub struct HyperliquidClient { base_url: BaseUrl, @@ -65,30 +65,10 @@ impl ExchangeClient for HyperliquidClient { } async fn get_wallet_holdings(&self) -> anyhow::Result> { - let address = self.wallet_address; - let user_state = self.info_client.user_state(address).await?; - - Ok(user_state - .asset_positions - .into_iter() - .map(|asset_position| { - let position = asset_position.position; - - WalletHolding { - coin: position.coin, - total: position.szi, - entry_px: position.entry_px, - leverage: Some(position.leverage.value.to_string()), - unrealized_pnl: Some(position.unrealized_pnl), - realised_pnl: None, - funding_unlocked: Some(position.cum_funding.since_open), - collateral: Some(position.margin_used), - } - }) - .collect()) + balances::get_wallet_holdings(&self.info_client, self.wallet_address).await } async fn place_order(&self, order: Order) -> anyhow::Result { - Ok(order) + orders::place_order(self.exchange_client.as_ref(), order).await } } diff --git a/src/exchanges/hyperliquid/market_data.rs b/src/exchanges/hyperliquid/market_data.rs index 34a6b13..34c0bfb 100644 --- a/src/exchanges/hyperliquid/market_data.rs +++ b/src/exchanges/hyperliquid/market_data.rs @@ -5,6 +5,7 @@ use tokio::{ sync::{mpsc::unbounded_channel, watch::Sender}, time::Instant, }; +use tracing::{debug, info}; use crate::market_data::{MarketSnapshot, MarketState}; @@ -43,10 +44,20 @@ pub async fn subscribe( symbols: Vec, market_tx: Sender, ) -> anyhow::Result<()> { + info!( + symbol_count = symbols.len(), + "opening hyperliquid market data stream" + ); + let mut info_client = InfoClient::new(None, Some(base_url)).await?; let (sender, mut receiver) = unbounded_channel(); for symbol in symbols { + info!( + symbol = %symbol, + "subscribing to hyperliquid l2 book" + ); + info_client .subscribe(Subscription::L2Book { coin: symbol }, sender.clone()) .await?; @@ -63,11 +74,19 @@ pub async fn subscribe( market_state.update(snapshot_from_l2_data(&l2_book.data)?); + debug!( + symbol = %l2_book.data.coin, + "received hyperliquid l2 book update" + ); + if market_tx.send(market_state.clone()).is_err() { + info!("market data receiver dropped"); break; } } + info!("hyperliquid market data stream closed"); + Ok(()) } diff --git a/src/exchanges/hyperliquid/mod.rs b/src/exchanges/hyperliquid/mod.rs index 3e4121c..5da19e5 100644 --- a/src/exchanges/hyperliquid/mod.rs +++ b/src/exchanges/hyperliquid/mod.rs @@ -1,4 +1,6 @@ +mod balances; mod client; mod market_data; +mod orders; pub use client::HyperliquidClient; diff --git a/src/exchanges/hyperliquid/orders.rs b/src/exchanges/hyperliquid/orders.rs new file mode 100644 index 0000000..cbb2a8a --- /dev/null +++ b/src/exchanges/hyperliquid/orders.rs @@ -0,0 +1,120 @@ +use hyperliquid_rust_sdk::{ + ClientLimit, ClientOrder, ClientOrderRequest, ExchangeClient as HlExchangeClient, + ExchangeDataStatus, ExchangeResponseStatus, +}; +use tracing::{info, warn}; + +use crate::order_state::{Order, OrderStatus, Side}; + +pub async fn place_order( + exchange_client: Option<&HlExchangeClient>, + order: Order, +) -> anyhow::Result { + let exchange_client = exchange_client + .ok_or_else(|| anyhow::anyhow!("order placement is disabled; set ALLOW_ORDER=true"))?; + + if order.price <= 0.0 { + anyhow::bail!("order price must be positive"); + } + + if order.size <= 0.0 { + anyhow::bail!("order size must be positive"); + } + + info!( + symbol = %order.symbol, + order_id = order.order_id, + side = side_as_str(&order.side), + price = order.price, + size = order.size, + "sending order to hyperliquid" + ); + + let response = exchange_client + .order( + ClientOrderRequest { + asset: order.symbol.clone(), + is_buy: order.side == Side::Buy, + reduce_only: false, + limit_px: order.price, + sz: order.size, + cloid: None, + order_type: ClientOrder::Limit(ClientLimit { + tif: "Gtc".to_string(), + }), + }, + None, + ) + .await?; + + let submitted_order = order_from_exchange_response(order, response)?; + + info!( + symbol = %submitted_order.symbol, + order_id = submitted_order.order_id, + status = ?submitted_order.status, + "hyperliquid order response received" + ); + + Ok(submitted_order) +} + +fn order_from_exchange_response( + mut order: Order, + response: ExchangeResponseStatus, +) -> anyhow::Result { + match response { + ExchangeResponseStatus::Ok(response) => { + let status = response + .data + .and_then(|data| data.statuses.into_iter().next()) + .unwrap_or(ExchangeDataStatus::Success); + + match status { + ExchangeDataStatus::Success => { + order.status = OrderStatus::New; + } + ExchangeDataStatus::WaitingForFill | ExchangeDataStatus::WaitingForTrigger => { + order.status = OrderStatus::New; + } + ExchangeDataStatus::Resting(resting) => { + order.order_id = resting.oid; + order.status = OrderStatus::New; + } + ExchangeDataStatus::Filled(filled) => { + order.order_id = filled.oid; + order.status = OrderStatus::Filled; + + if let Ok(avg_px) = filled.avg_px.parse::() { + order.price = avg_px; + } + + if let Ok(total_sz) = filled.total_sz.parse::() { + order.size = total_sz; + } + } + ExchangeDataStatus::Error(reason) => { + warn!( + symbol = %order.symbol, + order_id = order.order_id, + reason, + "hyperliquid rejected order" + ); + order.status = OrderStatus::Rejected { reason }; + } + } + + Ok(order) + } + ExchangeResponseStatus::Err(reason) => { + anyhow::bail!("hyperliquid order request failed: {reason}"); + } + } +} + +fn side_as_str(side: &Side) -> &'static str { + match side { + Side::Buy => "buy", + Side::Sell => "sell", + } +} diff --git a/src/execution.rs b/src/execution.rs index 23f579f..619e4bc 100644 --- a/src/execution.rs +++ b/src/execution.rs @@ -1,18 +1,102 @@ use crate::{ - config::Config, exchanges::ExchangeClient, market_data::MarketState, order_state::OrderState, + config::Config, + exchanges::ExchangeClient, + market_data::MarketState, + order_state::{OrderState, Side}, + risk::validate_order, }; use std::sync::Arc; use tokio::sync::watch::Receiver; +use tracing::{error, info, warn}; pub async fn run_loop( config: Arc, exchange: Arc, market_rx: Receiver, - order_rx: Receiver, + mut order_rx: Receiver, ) -> anyhow::Result<()> { - // , order_state: OrderState - loop { - // TODO: Check for new proposed orders from order_state, validate with risk::validate_order, and execute if valid - tokio::time::sleep(std::time::Duration::from_secs(1)).await; + info!("execution loop starting"); + + while order_rx.changed().await.is_ok() { + let Some(order) = order_rx.borrow_and_update().proposed_order.clone() else { + continue; + }; + + let side = side_as_str(&order.side); + let market = market_rx.borrow().get(&order.symbol).cloned(); + let Some(market) = market else { + warn!( + symbol = %order.symbol, + order_id = order.order_id, + side, + price = order.price, + size = order.size, + "rejecting order because market data is unavailable" + ); + continue; + }; + + if let Err(reject) = validate_order(&order, &market, &config) { + warn!( + symbol = %order.symbol, + order_id = order.order_id, + side, + price = order.price, + size = order.size, + bid = market.bid, + ask = market.ask, + mid = market.mid, + reject_reason = ?reject, + "order rejected by risk" + ); + continue; + } + + info!( + symbol = %order.symbol, + order_id = order.order_id, + side, + price = order.price, + size = order.size, + bid = market.bid, + ask = market.ask, + mid = market.mid, + "submitting order" + ); + + match exchange.place_order(order.clone()).await { + Ok(submitted_order) => { + info!( + symbol = %submitted_order.symbol, + order_id = submitted_order.order_id, + side = side_as_str(&submitted_order.side), + price = submitted_order.price, + size = submitted_order.size, + "order submitted" + ); + } + Err(err) => { + error!( + symbol = %order.symbol, + order_id = order.order_id, + side, + price = order.price, + size = order.size, + error = ?err, + "order submission failed" + ); + } + } + } + + info!("order state sender dropped; execution loop stopping"); + + Ok(()) +} + +fn side_as_str(side: &Side) -> &'static str { + match side { + Side::Buy => "buy", + Side::Sell => "sell", } } diff --git a/src/lib.rs b/src/lib.rs index 36c1944..f9fcea6 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -5,3 +5,4 @@ pub mod market_data; pub mod order_state; pub mod risk; pub mod wallet; +pub mod weights; diff --git a/src/main.rs b/src/main.rs index a2882da..78118a9 100644 --- a/src/main.rs +++ b/src/main.rs @@ -3,16 +3,25 @@ use hyperliquid_rust_exec_poc::exchanges; use hyperliquid_rust_exec_poc::execution; use hyperliquid_rust_exec_poc::market_data::{self, MarketState}; use hyperliquid_rust_exec_poc::order_state::{self, OrderState}; -// use crate::risk::validate_order use std::sync::Arc; use tokio::sync::watch::channel; +use tracing::info; +use tracing_subscriber::{EnvFilter, fmt}; + #[tokio::main] async fn main() -> anyhow::Result<()> { let config = Arc::new(Config::from_env()?); - // telemetry::init(&config)?; + init_tracing(&config); + + info!( + symbols = ?config.symbols, + allow_order = config.allow_order, + use_mainnet = config.use_mainnet, + "application starting" + ); let exchange = Arc::new(exchanges::hyperliquid::HyperliquidClient::new(&config).await?); let (market_tx, market_rx) = channel(MarketState::new()); @@ -23,9 +32,21 @@ async fn main() -> anyhow::Result<()> { Arc::clone(&exchange), market_tx, )); - tokio::spawn(order_state::run(Arc::clone(&config), order_tx)); + tokio::spawn(order_state::run( + Arc::clone(&config), + market_rx.clone(), + order_tx, + )); execution::run_loop(config, exchange, market_rx, order_rx).await?; Ok(()) } + +fn init_tracing(config: &Config) { + let filter = EnvFilter::try_from_default_env() + .or_else(|_| EnvFilter::try_new(&config.log_level)) + .unwrap_or_else(|_| EnvFilter::new("info")); + + fmt().with_env_filter(filter).json().init(); +} diff --git a/src/market_data.rs b/src/market_data.rs index 320571d..4e696c6 100644 --- a/src/market_data.rs +++ b/src/market_data.rs @@ -3,6 +3,7 @@ use std::collections::HashMap; use std::sync::Arc; use tokio::{sync::watch::Sender, time::Instant}; +use tracing::{error, info}; use crate::{config::Config, exchanges::ExchangeClient}; @@ -40,10 +41,18 @@ pub async fn run(config: Arc, exchange: Arc, market_tx: Sender, order_tx: tokio::sync::watch::Sender) { - // TODO: Connect to Hyperliquid WebSocket to receive order updates and update order_state.proposed_order accordingly +pub async fn run( + config: Arc, + market_rx: Receiver, + order_tx: Sender, +) { + let mut interval = tokio::time::interval(std::time::Duration::from_secs(5)); + + loop { + interval.tick().await; + + let weights = match crate::weights::get_target_weights(&config).await { + Ok(weights) => weights, + Err(err) => { + warn!( + error = ?err, + "failed to retrieve target weights" + ); + continue; + } + }; + + debug!( + weights = ?weights, + "retrieved target weights" + ); + + let market_state = market_rx.borrow().clone(); + let orders = generate_orders(&config, &market_state, &weights); + + if orders.is_empty() { + warn!("no orders generated from target weights"); + continue; + } + + for order in orders { + info!( + symbol = %order.symbol, + order_id = order.order_id, + side = side_as_str(&order.side), + price = order.price, + size = order.size, + "publishing proposed order" + ); + + if order_tx + .send(OrderState { + proposed_order: Some(order), + }) + .is_err() + { + info!("order receiver dropped; stopping order producer"); + return; + } + + tokio::time::sleep(std::time::Duration::from_millis(100)).await; + } + } +} + +fn generate_orders( + config: &Config, + market_state: &MarketState, + weights: &SymbolWeights, +) -> Vec { + weights + .iter() + .filter_map(|(symbol, weight)| { + if *weight <= 0.0 { + return None; + } + + let market = match market_state.get(symbol) { + Some(market) => market, + None => { + warn!( + symbol = %symbol, + weight, + "skipping order because market snapshot is unavailable" + ); + return None; + } + }; + + let price = market.ask; + if price <= 0.0 { + warn!( + symbol = %symbol, + weight, + price, + "skipping order because market ask is unavailable" + ); + return None; + } + + let notional_usd = config.max_order_usd * weight; + let size = notional_usd / price; + + Some(Order { + status: OrderStatus::New, + symbol: symbol.clone(), + order_id: NEXT_ORDER_ID.fetch_add(1, Ordering::Relaxed), + side: Side::Buy, + price, + size, + }) + }) + .collect() +} + +fn side_as_str(side: &Side) -> &'static str { + match side { + Side::Buy => "buy", + Side::Sell => "sell", + } +} + +#[cfg(test)] +mod tests { + use std::collections::HashMap; + + use tokio::time::Instant; + + use super::*; + use crate::market_data::{MarketSnapshot, MarketState}; + + fn config(symbols: Vec<&str>) -> Config { + Config { + hl_base_url: "testnet".to_string(), + wallet_address: "0000000000000000000000000000000000000000".to_string(), + private_key: None, + symbols: symbols.into_iter().map(str::to_string).collect(), + allow_order: false, + use_mainnet: false, + max_order_usd: 100.0, + max_position_usd: 500.0, + quote_asset: "USDC".to_string(), + log_level: "info".to_string(), + } + } + + fn market_snapshot(symbol: &str, bid: f64, ask: f64) -> MarketSnapshot { + MarketSnapshot { + symbol: symbol.to_string(), + bid, + ask, + mid: (bid + ask) / 2.0, + received_at: Instant::now(), + } + } + + #[test] + fn generate_orders_sizes_orders_from_weights_and_market_asks() { + let config = config(vec!["ETH", "BTC"]); + let mut market_state = MarketState::new(); + market_state.update(market_snapshot("ETH", 99.0, 100.0)); + market_state.update(market_snapshot("BTC", 199.0, 200.0)); + + let weights = HashMap::from([("ETH".to_string(), 0.25), ("BTC".to_string(), 0.75)]); + + let mut orders = generate_orders(&config, &market_state, &weights); + orders.sort_by(|left, right| left.symbol.cmp(&right.symbol)); + + assert_eq!(orders.len(), 2); + + let btc = &orders[0]; + assert_eq!(btc.symbol, "BTC"); + assert_eq!(btc.side, Side::Buy); + assert_eq!(btc.price, 200.0); + assert_eq!(btc.size, 0.375); + + let eth = &orders[1]; + assert_eq!(eth.symbol, "ETH"); + assert_eq!(eth.side, Side::Buy); + assert_eq!(eth.price, 100.0); + assert_eq!(eth.size, 0.25); + } + + #[test] + fn generate_orders_skips_symbols_without_market_data() { + let config = config(vec!["ETH", "BTC"]); + let mut market_state = MarketState::new(); + market_state.update(market_snapshot("ETH", 99.0, 100.0)); + + let weights = HashMap::from([("ETH".to_string(), 0.4), ("BTC".to_string(), 0.6)]); + + let orders = generate_orders(&config, &market_state, &weights); + + assert_eq!(orders.len(), 1); + assert_eq!(orders[0].symbol, "ETH"); + assert_eq!(orders[0].size, 0.4); + } + + #[test] + fn generate_orders_skips_zero_or_negative_weights() { + let config = config(vec!["ETH", "BTC"]); + let mut market_state = MarketState::new(); + market_state.update(market_snapshot("ETH", 99.0, 100.0)); + market_state.update(market_snapshot("BTC", 199.0, 200.0)); + + let weights = HashMap::from([("ETH".to_string(), 0.0), ("BTC".to_string(), -0.25)]); + + let orders = generate_orders(&config, &market_state, &weights); + + assert!(orders.is_empty()); + } } diff --git a/src/risk.rs b/src/risk.rs index 5e9fc56..b2af73b 100644 --- a/src/risk.rs +++ b/src/risk.rs @@ -19,6 +19,7 @@ pub fn validate_order( Ok(()) } +#[derive(Debug)] pub enum RiskReject { OrderTooLarge, StaleMarketData, diff --git a/src/weights.rs b/src/weights.rs new file mode 100644 index 0000000..854ed3b --- /dev/null +++ b/src/weights.rs @@ -0,0 +1,33 @@ +use std::collections::HashMap; + +use rand::Rng; + +use crate::config::Config; + +pub type SymbolWeights = HashMap; + +pub async fn get_target_weights(config: &Config) -> anyhow::Result { + if config.symbols.is_empty() { + anyhow::bail!("SYMBOLS must contain at least one symbol"); + } + + let raw_weights = { + let mut rng = rand::thread_rng(); + config + .symbols + .iter() + .map(|symbol| (symbol.clone(), rng.gen_range(0.0..1.0))) + .collect::>() + }; + + let total = raw_weights.iter().map(|(_, weight)| weight).sum::(); + + if total <= 0.0 { + anyhow::bail!("generated target weights have zero total"); + } + + Ok(raw_weights + .into_iter() + .map(|(symbol, weight)| (symbol, weight / total)) + .collect()) +} From 1cb846c6afc9de2db3a0397c48f49e4c1af74181 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Thu, 14 May 2026 15:52:37 +0000 Subject: [PATCH 2/8] switch order channel from watch to mpsc to prevent dropped orders Agent-Logs-Url: https://github.com/moconnell/rust-exec/sessions/0d3a9f04-aa3e-41ed-9476-d118ea9f10a7 Co-authored-by: moconnell <640958+moconnell@users.noreply.github.com> --- src/execution.rs | 14 +++++--------- src/main.rs | 8 ++++---- src/order_state.rs | 28 ++++------------------------ 3 files changed, 13 insertions(+), 37 deletions(-) diff --git a/src/execution.rs b/src/execution.rs index 619e4bc..960958d 100644 --- a/src/execution.rs +++ b/src/execution.rs @@ -2,26 +2,22 @@ use crate::{ config::Config, exchanges::ExchangeClient, market_data::MarketState, - order_state::{OrderState, Side}, + order_state::{Order, Side}, risk::validate_order, }; use std::sync::Arc; -use tokio::sync::watch::Receiver; +use tokio::sync::{mpsc, watch::Receiver}; use tracing::{error, info, warn}; pub async fn run_loop( config: Arc, exchange: Arc, market_rx: Receiver, - mut order_rx: Receiver, + mut order_rx: mpsc::Receiver, ) -> anyhow::Result<()> { info!("execution loop starting"); - while order_rx.changed().await.is_ok() { - let Some(order) = order_rx.borrow_and_update().proposed_order.clone() else { - continue; - }; - + while let Some(order) = order_rx.recv().await { let side = side_as_str(&order.side); let market = market_rx.borrow().get(&order.symbol).cloned(); let Some(market) = market else { @@ -89,7 +85,7 @@ pub async fn run_loop( } } - info!("order state sender dropped; execution loop stopping"); + info!("order sender dropped; execution loop stopping"); Ok(()) } diff --git a/src/main.rs b/src/main.rs index 78118a9..bc1f758 100644 --- a/src/main.rs +++ b/src/main.rs @@ -2,11 +2,11 @@ use hyperliquid_rust_exec_poc::config::Config; use hyperliquid_rust_exec_poc::exchanges; use hyperliquid_rust_exec_poc::execution; use hyperliquid_rust_exec_poc::market_data::{self, MarketState}; -use hyperliquid_rust_exec_poc::order_state::{self, OrderState}; +use hyperliquid_rust_exec_poc::order_state; use std::sync::Arc; -use tokio::sync::watch::channel; +use tokio::sync::{mpsc, watch}; use tracing::info; use tracing_subscriber::{EnvFilter, fmt}; @@ -24,8 +24,8 @@ async fn main() -> anyhow::Result<()> { ); let exchange = Arc::new(exchanges::hyperliquid::HyperliquidClient::new(&config).await?); - let (market_tx, market_rx) = channel(MarketState::new()); - let (order_tx, order_rx) = channel(OrderState::new()); + let (market_tx, market_rx) = watch::channel(MarketState::new()); + let (order_tx, order_rx) = mpsc::channel(1024); tokio::spawn(market_data::run( Arc::clone(&config), diff --git a/src/order_state.rs b/src/order_state.rs index ace5f9f..7159bbb 100644 --- a/src/order_state.rs +++ b/src/order_state.rs @@ -2,7 +2,8 @@ use std::clone; use std::sync::Arc; use std::sync::atomic::{AtomicU64, Ordering}; -use tokio::sync::watch::{Receiver, Sender}; +use tokio::sync::mpsc; +use tokio::sync::watch::Receiver; use tracing::{debug, info, warn}; use crate::{config::Config, market_data::MarketState, weights::SymbolWeights}; @@ -57,24 +58,10 @@ pub enum Side { Sell, } -#[derive(clone::Clone)] -pub struct OrderState { - // For simplicity, we just track the latest proposed order. In a real implementation, this would likely be a queue or more complex structure. - pub proposed_order: Option, -} - -impl OrderState { - pub fn new() -> Self { - Self { - proposed_order: None, - } - } -} - pub async fn run( config: Arc, market_rx: Receiver, - order_tx: Sender, + order_tx: mpsc::Sender, ) { let mut interval = tokio::time::interval(std::time::Duration::from_secs(5)); @@ -115,17 +102,10 @@ pub async fn run( "publishing proposed order" ); - if order_tx - .send(OrderState { - proposed_order: Some(order), - }) - .is_err() - { + if order_tx.send(order).await.is_err() { info!("order receiver dropped; stopping order producer"); return; } - - tokio::time::sleep(std::time::Duration::from_millis(100)).await; } } } From b24692de68adabb30527a814e3c2992db57b9a0b Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Thu, 14 May 2026 15:56:15 +0000 Subject: [PATCH 3/8] add named constant for order channel capacity with documentation Agent-Logs-Url: https://github.com/moconnell/rust-exec/sessions/0d3a9f04-aa3e-41ed-9476-d118ea9f10a7 Co-authored-by: moconnell <640958+moconnell@users.noreply.github.com> --- src/main.rs | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/src/main.rs b/src/main.rs index bc1f758..697f988 100644 --- a/src/main.rs +++ b/src/main.rs @@ -11,6 +11,10 @@ use tokio::sync::{mpsc, watch}; use tracing::info; use tracing_subscriber::{EnvFilter, fmt}; +/// Maximum number of orders that can be queued before the producer is back-pressured. +/// Sized to hold a full batch of orders (one per symbol) across several intervals. +const ORDER_CHANNEL_CAPACITY: usize = 1024; + #[tokio::main] async fn main() -> anyhow::Result<()> { let config = Arc::new(Config::from_env()?); @@ -25,7 +29,7 @@ async fn main() -> anyhow::Result<()> { let exchange = Arc::new(exchanges::hyperliquid::HyperliquidClient::new(&config).await?); let (market_tx, market_rx) = watch::channel(MarketState::new()); - let (order_tx, order_rx) = mpsc::channel(1024); + let (order_tx, order_rx) = mpsc::channel(ORDER_CHANNEL_CAPACITY); tokio::spawn(market_data::run( Arc::clone(&config), From 616c6322a643d4a202fc314a1ca36ed0b48a47c2 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Thu, 14 May 2026 16:00:45 +0000 Subject: [PATCH 4/8] fix duplicate symbols and add dry-run mode short-circuit Agent-Logs-Url: https://github.com/moconnell/rust-exec/sessions/2a4a6e61-d9e6-4b4a-a7b1-de5cdb2fb892 Co-authored-by: moconnell <640958+moconnell@users.noreply.github.com> --- src/execution.rs | 21 ++++++++++++++++++++- src/weights.rs | 21 +++++++++++---------- 2 files changed, 31 insertions(+), 11 deletions(-) diff --git a/src/execution.rs b/src/execution.rs index 960958d..57ef616 100644 --- a/src/execution.rs +++ b/src/execution.rs @@ -15,7 +15,10 @@ pub async fn run_loop( market_rx: Receiver, mut order_rx: mpsc::Receiver, ) -> anyhow::Result<()> { - info!("execution loop starting"); + info!( + allow_order = config.allow_order, + "execution loop starting" + ); while let Some(order) = order_rx.recv().await { let side = side_as_str(&order.side); @@ -48,6 +51,22 @@ pub async fn run_loop( continue; } + // Short-circuit in dry-run mode to avoid spamming error logs + if !config.allow_order { + info!( + symbol = %order.symbol, + order_id = order.order_id, + side, + price = order.price, + size = order.size, + bid = market.bid, + ask = market.ask, + mid = market.mid, + "dry-run: order validated but not submitted (allow_order=false)" + ); + continue; + } + info!( symbol = %order.symbol, order_id = order.order_id, diff --git a/src/weights.rs b/src/weights.rs index 854ed3b..e4c0028 100644 --- a/src/weights.rs +++ b/src/weights.rs @@ -11,16 +11,17 @@ pub async fn get_target_weights(config: &Config) -> anyhow::Result>() - }; - - let total = raw_weights.iter().map(|(_, weight)| weight).sum::(); + // Deduplicate symbols and generate weights in a single pass to ensure + // the total matches what will be returned in the HashMap + let mut raw_weights = HashMap::new(); + let mut rng = rand::thread_rng(); + + for symbol in &config.symbols { + // If symbol already exists, this will overwrite it, ensuring deduplication + raw_weights.insert(symbol.clone(), rng.gen_range(0.0..1.0)); + } + + let total = raw_weights.values().sum::(); if total <= 0.0 { anyhow::bail!("generated target weights have zero total"); From 477d86e14b5c337a4136a01f1d40be45b05776a0 Mon Sep 17 00:00:00 2001 From: Matthew O'Connell Date: Thu, 14 May 2026 18:08:41 +0200 Subject: [PATCH 5/8] update README --- README.md | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/README.md b/README.md index 8a20c67..93af89b 100644 --- a/README.md +++ b/README.md @@ -1,13 +1,14 @@ # rust-exec -A proof-of-concept to explore exchange connectivity with [Hyperliquid](hyperliquid.xyz) and trade workflow using the Rust language, [tokio](https://github.com/tokio-rs/tokio) and [hyperliquid_rust_sdk](https://github.com/hyperliquid-dex/hyperliquid-rust-sdk). +A proof-of-concept Rust trade execution later, to explore exchange connectivity and trade workflow with [Hyperliquid](hyperliquid.xyz) using [tokio](https://github.com/tokio-rs/tokio) and [hyperliquid_rust_sdk](https://github.com/hyperliquid-dex/hyperliquid-rust-sdk). ## Features - Rust/Tokio async exchange connectivity - Hyperliquid L2 market data ingestion - exchange abstraction via trait -- shared state propagation using watch channels +- shared state propagation using mpsc & watch channels +- dummy weights collection - dry-run execution loop - pre-trade risk check framework - testnet integration tests From 11ffd96723cbdabd90fe46f0a01529249d7351c5 Mon Sep 17 00:00:00 2001 From: Matthew O'Connell Date: Thu, 14 May 2026 18:11:49 +0200 Subject: [PATCH 6/8] fix: README spelling --- README.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/README.md b/README.md index 93af89b..e725a06 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,6 @@ # rust-exec -A proof-of-concept Rust trade execution later, to explore exchange connectivity and trade workflow with [Hyperliquid](hyperliquid.xyz) using [tokio](https://github.com/tokio-rs/tokio) and [hyperliquid_rust_sdk](https://github.com/hyperliquid-dex/hyperliquid-rust-sdk). +A proof-of-concept trade execution layer in Rust, exploring exchange connectivity and trade workflow with [Hyperliquid](hyperliquid.xyz) using [tokio](https://github.com/tokio-rs/tokio) and [hyperliquid_rust_sdk](https://github.com/hyperliquid-dex/hyperliquid-rust-sdk). ## Features From 9855a8d59200d223d99d390ec7d248773ba17134 Mon Sep 17 00:00:00 2001 From: Matthew O'Connell Date: Thu, 14 May 2026 18:12:42 +0200 Subject: [PATCH 7/8] refactor: move .env.example to root dir --- src/.env.example => .env.example | 0 1 file changed, 0 insertions(+), 0 deletions(-) rename src/.env.example => .env.example (100%) diff --git a/src/.env.example b/.env.example similarity index 100% rename from src/.env.example rename to .env.example From e6c7fe20829d14c2fe523dc356351a0a5a6462ac Mon Sep 17 00:00:00 2001 From: Matthew O'Connell Date: Thu, 14 May 2026 18:15:49 +0200 Subject: [PATCH 8/8] refactor: remove HL_BASE_URL (unused) --- .env.example | 1 - src/config.rs | 3 --- src/order_state.rs | 1 - tests/hyperliquid_market_data.rs | 1 - 4 files changed, 6 deletions(-) diff --git a/.env.example b/.env.example index 19fe38e..dda6847 100644 --- a/.env.example +++ b/.env.example @@ -1,4 +1,3 @@ -HL_BASE_URL=https://api.hyperliquid.xyz WALLET_ADDRESS=0000000000000000000000000000000000000000 PRIVATE_KEY= diff --git a/src/config.rs b/src/config.rs index f6ed20c..7a977c0 100644 --- a/src/config.rs +++ b/src/config.rs @@ -2,7 +2,6 @@ use std::clone; #[derive(clone::Clone)] pub struct Config { - pub hl_base_url: String, pub wallet_address: String, pub private_key: Option, @@ -19,7 +18,6 @@ pub struct Config { impl Config { pub fn from_env() -> anyhow::Result { - let hl_base_url = std::env::var("HL_BASE_URL")?; let wallet_address = std::env::var("WALLET_ADDRESS")?; let private_key = std::env::var("PRIVATE_KEY").ok(); @@ -45,7 +43,6 @@ impl Config { let log_level = std::env::var("LOG_LEVEL").unwrap_or_else(|_| "info".to_string()); Ok(Self { - hl_base_url, wallet_address, private_key, symbols, diff --git a/src/order_state.rs b/src/order_state.rs index 7159bbb..bb7ee35 100644 --- a/src/order_state.rs +++ b/src/order_state.rs @@ -178,7 +178,6 @@ mod tests { fn config(symbols: Vec<&str>) -> Config { Config { - hl_base_url: "testnet".to_string(), wallet_address: "0000000000000000000000000000000000000000".to_string(), private_key: None, symbols: symbols.into_iter().map(str::to_string).collect(), diff --git a/tests/hyperliquid_market_data.rs b/tests/hyperliquid_market_data.rs index 9e85cff..b76ad79 100644 --- a/tests/hyperliquid_market_data.rs +++ b/tests/hyperliquid_market_data.rs @@ -9,7 +9,6 @@ use tokio::{sync::watch, time::timeout}; fn test_config(symbols: Vec) -> Arc { Arc::new(Config { - hl_base_url: "testnet".to_string(), wallet_address: "0000000000000000000000000000000000000000".to_string(), private_key: None, symbols,