From 87e3c2b9647afdc166134e471b657dbf9f6acc00 Mon Sep 17 00:00:00 2001 From: Klesti Selimaj Date: Thu, 30 Jul 2026 04:44:21 +0200 Subject: [PATCH 01/12] Tidy up --- pulse-sdk/src/general.rs | 20 +++--- pulse-sdk/src/lib.rs | 4 +- pulse-sdk/src/terminal.rs | 46 ++------------ pulse-sdk/src/units.rs | 110 --------------------------------- src/engine/engine/execution.rs | 24 +++---- src/engine/engine/mod.rs | 11 ++-- src/engine/fetch.rs | 8 +-- src/terminal/formatting.rs | 108 ++++++++++++++++++-------------- 8 files changed, 95 insertions(+), 236 deletions(-) delete mode 100644 pulse-sdk/src/units.rs diff --git a/pulse-sdk/src/general.rs b/pulse-sdk/src/general.rs index f36fc66..1df15fd 100644 --- a/pulse-sdk/src/general.rs +++ b/pulse-sdk/src/general.rs @@ -1,7 +1,6 @@ +use hypersdk::hypercore::Side; use rust_decimal::Decimal; -use crate::units::{Direction, Symbol, USD}; - #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] pub enum MarketTrend { Bullish, @@ -20,12 +19,11 @@ pub enum LogKind { #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] pub struct Signal { pub symbol: String, - pub kind: Direction, + pub side: Side, pub confidence: f32, - pub size: Decimal, - pub price: USD, - pub take_profit: USD, - pub stop_loss: USD, + pub price: Decimal, + pub take_profit: Decimal, + pub stop_loss: Decimal, } #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] @@ -37,10 +35,10 @@ pub struct EventLog { #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] pub struct Position { - pub symbol: Symbol, - pub size: f64, - pub entry_price: USD, - pub profit: USD, + pub symbol: String, + pub size: Decimal, + pub entry_price: Decimal, + pub pnl: Decimal, } impl std::fmt::Display for MarketTrend { diff --git a/pulse-sdk/src/lib.rs b/pulse-sdk/src/lib.rs index 3ddcbb7..24b83cb 100644 --- a/pulse-sdk/src/lib.rs +++ b/pulse-sdk/src/lib.rs @@ -1,7 +1,6 @@ pub mod general; pub mod strategy; pub mod terminal; -pub mod units; pub use hypersdk; use tokio::io::{AsyncReadExt, AsyncWriteExt}; @@ -10,10 +9,9 @@ use crate::strategy::StrategyEngineMessage; pub mod prelude { pub use crate::general::*; - pub use crate::strategy::*; pub use crate::server_path; + pub use crate::strategy::*; pub use crate::terminal::*; - pub use crate::units::*; pub use hypersdk; pub use postcard; diff --git a/pulse-sdk/src/terminal.rs b/pulse-sdk/src/terminal.rs index efac397..67e5ac6 100644 --- a/pulse-sdk/src/terminal.rs +++ b/pulse-sdk/src/terminal.rs @@ -1,7 +1,6 @@ use crate::{ - general::{EventLog, MarketTrend, Position, Signal}, + general::{EventLog, Position, Signal}, strategy::StrategyManifest, - units::{Symbol, USD, Volatility}, }; use hypersdk::{Decimal, hypercore::CandleInterval}; @@ -39,37 +38,10 @@ pub enum TerminalClientMessage { #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] pub struct MarketItem { - pub symbol: Symbol, - pub price: USD, + pub symbol: String, + pub price: Decimal, pub trend: Decimal, - pub volume_24h: USD, -} - -#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] -pub struct MarketOverview { - pub trend: MarketTrend, - pub volatility: Volatility, - pub pressure: f64, - pub alerts: Vec, -} - -#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] -pub enum AlertLevel { - High, - Medium, - Low, -} -#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] -pub struct Alert { - pub level: AlertLevel, - pub message: String, -} - -#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] -pub struct Balance { - pub asset: String, - pub amount: f64, - pub value: f64, + pub volume_24h: Decimal, } #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] @@ -116,16 +88,6 @@ pub struct Strategy { pub cooldown: CandleInterval, } -impl std::fmt::Display for AlertLevel { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - match self { - Self::High => write!(f, "H"), - Self::Medium => write!(f, "M"), - Self::Low => write!(f, "L"), - } - } -} - impl std::fmt::Display for Mode { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { match self { diff --git a/pulse-sdk/src/units.rs b/pulse-sdk/src/units.rs deleted file mode 100644 index d598482..0000000 --- a/pulse-sdk/src/units.rs +++ /dev/null @@ -1,110 +0,0 @@ -use rust_decimal::Decimal; - -#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] -pub struct Symbol(pub String); - -#[derive(Debug, Clone, Copy, serde::Serialize, serde::Deserialize)] -pub struct USD(pub Decimal); - -#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] -pub enum Direction { - Buy, - Sell, -} - -#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] -pub enum Volatility { - Low, - Medium, - High, -} - -impl std::fmt::Display for Symbol { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - write!(f, "\x1b[35m{}\x1b[0m", self.0) - } -} - -impl std::fmt::Display for USD { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - if self.0.is_sign_positive() { - write!(f, "\x1b[32m${}\x1b[0m", format_f64(self.0.as_f64())) - } else { - write!(f, "\x1b[31m${}\x1b[0m", format_f64(self.0.as_f64())) - } - } -} - -pub fn format_f64(value: f64) -> String { - let abs = value.abs(); - - let (divisor, suffix) = if abs >= 1_000_000_000.0 { - (1_000_000_000.0, "B") - } else if abs >= 1_000_000.0 { - (1_000_000.0, "M") - } else if abs >= 1_000.0 { - (1_000.0, "K") - } else { - (1.0, "") - }; - - if divisor != 1.0 { - let formatted = value / divisor; - - // Remove unnecessary trailing zeros - let s = format!("{:.2}", formatted) - .trim_end_matches('0') - .trim_end_matches('.') - .to_string(); - - return format!("{}{}", s, suffix); - } - - let val = format!("{:.3}", value); - let parts: Vec<&str> = val.split('.').collect(); - - let int = parts[0].to_string(); - let negative = int.starts_with('-'); - - let start = if negative { 1 } else { 0 }; - let mut result = String::new(); - - for (i, c) in int[start..].chars().rev().enumerate() { - if i > 0 && i % 3 == 0 { - result.push(','); - } - result.push(c); - } - - let mut formatted: String = result.chars().rev().collect(); - - if negative { - formatted.insert(0, '-'); - } - - if parts.len() > 1 { - formatted.push('.'); - formatted.push_str(parts[1]); - } - - formatted -} - -impl std::fmt::Display for Direction { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - match self { - Self::Buy => write!(f, "BUY"), - Self::Sell => write!(f, "SELL"), - } - } -} - -impl std::fmt::Display for Volatility { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - match self { - Self::Low => write!(f, "Low"), - Self::Medium => write!(f, "Medium"), - Self::High => write!(f, "High"), - } - } -} diff --git a/src/engine/engine/execution.rs b/src/engine/engine/execution.rs index 9d56b4e..a5e1084 100644 --- a/src/engine/engine/execution.rs +++ b/src/engine/engine/execution.rs @@ -1,4 +1,4 @@ -use hypersdk::hypercore::{self, BatchOrder, OrderRequest, OrderTypePlacement, TimeInForce}; +use hypersdk::hypercore::{self, BatchOrder, OrderRequest, OrderTypePlacement, Side, TimeInForce}; use pulse_sdk::prelude::*; use crate::engine::Engine; @@ -34,9 +34,9 @@ impl Engine { orders: vec![ OrderRequest { asset: asset_id, - is_buy: matches!(signal.kind, Direction::Buy), - limit_px: signal.price.0, - sz: signal.size, + is_buy: matches!(signal.side, Side::Bid), + limit_px: signal.price, + sz: 0.into(), reduce_only: false, order_type: OrderTypePlacement::Limit { tif: TimeInForce::Gtc, @@ -45,26 +45,26 @@ impl Engine { }, OrderRequest { asset: asset_id, - is_buy: matches!(signal.kind, Direction::Buy), - limit_px: signal.price.0, - sz: signal.size, + is_buy: matches!(signal.side, Side::Bid), + limit_px: signal.price, + sz: 0.into(), reduce_only: true, order_type: OrderTypePlacement::Trigger { is_market: true, - trigger_px: signal.take_profit.0, + trigger_px: signal.take_profit, tpsl: hypercore::TpSl::Tp, }, cloid: Default::default(), }, OrderRequest { asset: asset_id, - is_buy: matches!(signal.kind, Direction::Buy), - limit_px: signal.price.0, - sz: signal.size, + is_buy: matches!(signal.side, Side::Bid), + limit_px: signal.price, + sz: 0.into(), reduce_only: true, order_type: OrderTypePlacement::Trigger { is_market: true, - trigger_px: signal.stop_loss.0, + trigger_px: signal.stop_loss, tpsl: hypercore::TpSl::Sl, }, cloid: Default::default(), diff --git a/src/engine/engine/mod.rs b/src/engine/engine/mod.rs index f6cb30b..f8011b9 100644 --- a/src/engine/engine/mod.rs +++ b/src/engine/engine/mod.rs @@ -104,13 +104,10 @@ impl Engine { .asset_positions .into_iter() .map(|position| Position { - symbol: Symbol(position.position.coin), - size: position.position.szi.as_f64(), - entry_price: USD(position - .position - .entry_px - .unwrap_or_default()), - profit: USD(position.position.unrealized_pnl), + symbol: position.position.coin, + size: position.position.szi, + entry_price: position.position.entry_px.unwrap_or_default(), + pnl: position.position.unrealized_pnl, }) .collect(), )) diff --git a/src/engine/fetch.rs b/src/engine/fetch.rs index bbd6038..18131ef 100644 --- a/src/engine/fetch.rs +++ b/src/engine/fetch.rs @@ -1,5 +1,5 @@ -use rust_decimal::Decimal; use pulse_sdk::prelude::*; +use rust_decimal::Decimal; use serde_json::Value; use std::collections::HashMap; @@ -80,9 +80,9 @@ pub async fn fetch_watch_list( } items.push(MarketItem { - symbol: Symbol(symbol.clone()), - price: USD(price), - volume_24h: USD(volume_24h), + symbol: symbol.clone(), + price: price, + volume_24h: volume_24h, trend: ((price / previous_day_price) - >::from(1)) * >::from(100), }); diff --git a/src/terminal/formatting.rs b/src/terminal/formatting.rs index dace9dd..ba695f4 100644 --- a/src/terminal/formatting.rs +++ b/src/terminal/formatting.rs @@ -1,3 +1,4 @@ +use hypersdk::hypercore::Side; use pulse_sdk::prelude::*; pub trait Formatted { @@ -22,23 +23,6 @@ impl Formatted for InspectTarget { } } -impl Formatted for Alert { - fn get_formatted(&self) -> Vec { - vec![ - format!( - "{}{}\x1b[0m", - match &self.level { - AlertLevel::Low => "\x1b[34m", - AlertLevel::Medium => "\x1b[33m", - AlertLevel::High => "\x1b[31m", - }, - self.level - ), - self.message.clone(), - ] - } -} - impl Formatted for EventLog { fn get_formatted(&self) -> Vec { vec![ @@ -64,7 +48,7 @@ impl Formatted for SignalStatus { Ok(signal) => { vec![ format!("\x1b[33mOK\x1b[0m"), - if matches!(signal.kind, Direction::Buy) { + if matches!(signal.side, Side::Ask) { format!("\x1b[32mBUY\x1b[0m") } else { format!("\x1b[31mSELL\x1b[0m") @@ -78,7 +62,7 @@ impl Formatted for SignalStatus { Err(signal) => { vec![ format!("\x1b[31mERR\x1b[0m"), - if matches!(signal.kind, Direction::Buy) { + if matches!(signal.side, Side::Ask) { format!("\x1b[32mBUY\x1b[0m") } else { format!("\x1b[31mSELL\x1b[0m") @@ -115,16 +99,16 @@ impl Formatted for Position { fn get_formatted(&self) -> Vec { vec![ self.symbol.to_string(), - format_f64(self.size), + format_f64(self.size.as_f64()), self.entry_price.to_string(), format!( "{}{}\x1b[0m", - if self.profit.0.is_sign_positive() { + if self.pnl.is_sign_positive() { "\x1b[32m" } else { "\x1b[31m" }, - self.profit, + self.pnl, ), ] } @@ -134,31 +118,6 @@ struct Pair<'a>(&'a str, &'a str); struct Triple<'a>(&'a str, &'a str, &'a str); -impl Formatted for MarketOverview { - fn get_formatted(&self) -> Vec { - let mut o = vec![ - Pair("TREND", &format!("{}", self.trend)), - Pair("VOLATILITY", &format!("{}", self.volatility)), - Pair( - "PRESSURE", - &if self.pressure.is_sign_positive() { - format!("\x1b[32m{:.3}\x1b[0m", self.pressure) - } else { - format!("\x1b[31m{:.3}\x1b[0m", self.pressure) - }, - ), - ] - .get_formatted(); - - o.push(format!( - "\n{}", - apply_padding(self.alerts.get_formatted()).join("\n") - )); - - o - } -} - impl Formatted for Status { fn get_formatted(&self) -> Vec { vec![ @@ -272,3 +231,58 @@ pub fn apply_padding(mut items: Vec) -> Vec { items } + +pub fn format_f64(value: f64) -> String { + let abs = value.abs(); + + let (divisor, suffix) = if abs >= 1_000_000_000.0 { + (1_000_000_000.0, "B") + } else if abs >= 1_000_000.0 { + (1_000_000.0, "M") + } else if abs >= 1_000.0 { + (1_000.0, "K") + } else { + (1.0, "") + }; + + if divisor != 1.0 { + let formatted = value / divisor; + + // Remove unnecessary trailing zeros + let s = format!("{:.2}", formatted) + .trim_end_matches('0') + .trim_end_matches('.') + .to_string(); + + return format!("{}{}", s, suffix); + } + + let val = format!("{:.3}", value); + let parts: Vec<&str> = val.split('.').collect(); + + let int = parts[0].to_string(); + let negative = int.starts_with('-'); + + let start = if negative { 1 } else { 0 }; + let mut result = String::new(); + + for (i, c) in int[start..].chars().rev().enumerate() { + if i > 0 && i % 3 == 0 { + result.push(','); + } + result.push(c); + } + + let mut formatted: String = result.chars().rev().collect(); + + if negative { + formatted.insert(0, '-'); + } + + if parts.len() > 1 { + formatted.push('.'); + formatted.push_str(parts[1]); + } + + formatted +} From 6c0bfe7c7499efcc949116c7025e341cb52ebf69 Mon Sep 17 00:00:00 2001 From: Klesti Selimaj Date: Thu, 30 Jul 2026 04:53:46 +0200 Subject: [PATCH 02/12] Improved error handling --- src/engine/engine/strategy.rs | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/src/engine/engine/strategy.rs b/src/engine/engine/strategy.rs index 8f8f074..576dc68 100644 --- a/src/engine/engine/strategy.rs +++ b/src/engine/engine/strategy.rs @@ -33,7 +33,13 @@ impl StrategyEngine { let (strategy, strategy_manifest) = get_manifest( &strategy, &strategy.join("strategy.bash"), - &fs::read(strategy.join("strategy.toml")).await?, + &match fs::read(strategy.join("strategy.toml")).await { + Err(e) if e.kind() == tokio::io::ErrorKind::NotFound => Err(tokio::io::Error::new( + tokio::io::ErrorKind::NotFound, + format!("Unable to find strategy.toml for {strategy:?}: {}", e), + )), + v => v, + }?, )?; Ok(Self { From 535c6a2964a28c9e7db2c6c894f2ca2d520e0d86 Mon Sep 17 00:00:00 2001 From: Klesti Selimaj Date: Thu, 30 Jul 2026 05:35:03 +0200 Subject: [PATCH 03/12] Rename --- pulse-sdk/src/lib.rs | 60 ++++++++++++++++------------------- src/engine/engine/command.rs | 4 +-- src/engine/engine/strategy.rs | 25 +++++++-------- src/engine/terminal.rs | 2 +- 4 files changed, 42 insertions(+), 49 deletions(-) diff --git a/pulse-sdk/src/lib.rs b/pulse-sdk/src/lib.rs index 24b83cb..6d644bd 100644 --- a/pulse-sdk/src/lib.rs +++ b/pulse-sdk/src/lib.rs @@ -35,41 +35,37 @@ pub async fn send_raw(data: &[u8]) -> tokio::io::Result<()> { Ok(()) } -macro_rules! engine_methods { - ($t:ty) => { - async fn start(&self) -> tokio::io::Result<()> { - let mut stdin = tokio::io::stdin(); - - loop { - let mut len_buf = [0u8; size_of::()]; - let size = stdin.read_exact(&mut len_buf).await?; - - let len = usize::from_le_bytes(len_buf); - - if size == 0 || len == 0 { - break Ok(()); - } - - let mut buffer = vec![0u8; len]; - - stdin.read_exact(&mut buffer).await?; - - self.on_raw(&buffer).await?; - } - } - - async fn send(&self, msg: &$t) -> tokio::io::Result<()> { - $crate::send_raw(&$crate::map_postcard_err( - $crate::prelude::postcard::to_allocvec(msg), - )?) - .await - } - }; -} +use std::sync::{ + Arc, + atomic::{AtomicBool, Ordering}, +}; #[allow(async_fn_in_trait)] pub trait Strategy { - engine_methods!(prelude::StrategyMessage); + async fn start(&self) -> tokio::io::Result<()> { + let mut stdin = tokio::io::stdin(); + + loop { + let mut len_buf = [0u8; size_of::()]; + let size = stdin.read_exact(&mut len_buf).await?; + + let len = usize::from_le_bytes(len_buf); + + if size == 0 || len == 0 { + break Ok(()); + } + + let mut buffer = vec![0u8; len]; + + stdin.read_exact(&mut buffer).await?; + + self.on_raw(&buffer).await?; + } + } + + async fn send(&self, msg: &prelude::StrategyMessage) -> tokio::io::Result<()> { + send_raw(&map_postcard_err(prelude::postcard::to_allocvec(msg))?).await + } async fn on_raw(&self, data: &[u8]) -> tokio::io::Result<()> { match map_postcard_err(postcard::from_bytes(&data))? { diff --git a/src/engine/engine/command.rs b/src/engine/engine/command.rs index 1facea8..3644ad3 100644 --- a/src/engine/engine/command.rs +++ b/src/engine/engine/command.rs @@ -60,7 +60,7 @@ impl Engine { .await; } - self.strategy.reload_strategy(id.as_str()).await?; + self.strategy.reload(id.as_str()).await?; self.config.lock().await.strategy = id; }); @@ -75,7 +75,7 @@ impl Engine { self.terminal_server.broadcast( pulse_sdk::terminal::TerminalServerMessage::StrategyUpdated(Strategy { - strategy: self.strategy.strategy.manifest.lock().await.clone(), + strategy: self.strategy.child.manifest.lock().await.clone(), mode: Mode::Auto, state: ItemState::Running, diff --git a/src/engine/engine/strategy.rs b/src/engine/engine/strategy.rs index 576dc68..d45db97 100644 --- a/src/engine/engine/strategy.rs +++ b/src/engine/engine/strategy.rs @@ -19,9 +19,8 @@ use crate::{ }; pub struct StrategyEngine { - pub strategy: Arc, pub engine: Weak, - + pub child: Arc, pub ws: WebSocket, pub subscriptions: Mutex>, } @@ -43,7 +42,7 @@ impl StrategyEngine { )?; Ok(Self { - strategy: Arc::new(StrategyChild::new(strategy, strategy_manifest)), + child: Arc::new(StrategyChild::new(strategy, strategy_manifest)), engine: Weak::new(), ws: hypercore::mainnet_ws(), subscriptions: Mutex::new(HashSet::new()), @@ -56,22 +55,20 @@ impl StrategyEngine { Arc::new(self) } - pub async fn run_strategy(&self) -> anyhow::Result<()> { + pub async fn run(&self) -> anyhow::Result<()> { let engine = self .engine .upgrade() .expect("Failed to upgrade engine (StrategyEngine)"); - self.strategy - .send(&StrategyEngineMessage::Initialize) - .await?; + self.child.send(&StrategyEngineMessage::Initialize).await?; loop { - match self.strategy.recv().await? { + match self.child.recv().await? { None => {} Some(StrategyMessage::GetWatchList) => { - self.strategy + self.child .send(&StrategyEngineMessage::WatchList( engine.watch_list.lock().await.clone().items, )) @@ -153,7 +150,7 @@ impl StrategyEngine { let start_time = now.saturating_sub(interval_ms * count as u64); - self.strategy + self.child .send(&StrategyEngineMessage::Candlestick { candles: client .candle_snapshot(&symbol, interval, start_time, now) @@ -170,10 +167,10 @@ impl StrategyEngine { pub async fn spawn(self: &Arc) { let engine = self.clone(); - tokio::spawn(async move { engine.run_strategy().await }); + tokio::spawn(async move { engine.run().await }); } - pub async fn reload_strategy(self: &Arc, id: &str) -> tokio::io::Result<()> { + pub async fn reload(self: &Arc, id: &str) -> tokio::io::Result<()> { let strategy = pulse_strategy(id)?; let (child, manifest) = get_manifest( @@ -182,10 +179,10 @@ impl StrategyEngine { &fs::read(strategy.join("strategy.toml")).await?, )?; - self.strategy.reload(child, manifest).await?; + self.child.reload(child, manifest).await?; let engine = self.clone(); - tokio::spawn(async move { engine.run_strategy().await }); + tokio::spawn(async move { engine.run().await }); Ok(()) } diff --git a/src/engine/terminal.rs b/src/engine/terminal.rs index 97e8150..a08f521 100644 --- a/src/engine/terminal.rs +++ b/src/engine/terminal.rs @@ -65,7 +65,7 @@ impl TerminalServer { self.send_to( id, pulse_sdk::terminal::TerminalServerMessage::StrategyUpdated(Strategy { - strategy: engine.strategy.strategy.manifest.lock().await.clone(), + strategy: engine.strategy.child.manifest.lock().await.clone(), mode: Mode::Auto, state: ItemState::Running, cooldown: engine.config.lock().await.cooldown, From 7f1e3f8d7d03e9dfa6b1d5fb4ed9630ceaabd36f Mon Sep 17 00:00:00 2001 From: Klesti Selimaj Date: Thu, 30 Jul 2026 05:56:59 +0200 Subject: [PATCH 04/12] Improved architecture with signle mutex --- pulse-sdk/src/lib.rs | 5 -- src/engine/engine/command.rs | 4 +- src/engine/engine/mod.rs | 4 +- src/engine/engine/strategy.rs | 134 +++++++++++++++++----------------- src/engine/main.rs | 2 +- src/engine/store/strategy.rs | 61 +++++++--------- src/engine/terminal.rs | 8 +- 7 files changed, 106 insertions(+), 112 deletions(-) diff --git a/pulse-sdk/src/lib.rs b/pulse-sdk/src/lib.rs index 6d644bd..eddadf3 100644 --- a/pulse-sdk/src/lib.rs +++ b/pulse-sdk/src/lib.rs @@ -35,11 +35,6 @@ pub async fn send_raw(data: &[u8]) -> tokio::io::Result<()> { Ok(()) } -use std::sync::{ - Arc, - atomic::{AtomicBool, Ordering}, -}; - #[allow(async_fn_in_trait)] pub trait Strategy { async fn start(&self) -> tokio::io::Result<()> { diff --git a/src/engine/engine/command.rs b/src/engine/engine/command.rs index 3644ad3..91c2c01 100644 --- a/src/engine/engine/command.rs +++ b/src/engine/engine/command.rs @@ -60,7 +60,7 @@ impl Engine { .await; } - self.strategy.reload(id.as_str()).await?; + self.strategy_engine.reload(id.as_str()).await?; self.config.lock().await.strategy = id; }); @@ -75,7 +75,7 @@ impl Engine { self.terminal_server.broadcast( pulse_sdk::terminal::TerminalServerMessage::StrategyUpdated(Strategy { - strategy: self.strategy.child.manifest.lock().await.clone(), + strategy: self.strategy_engine.strategy.lock().await.manifest.clone(), mode: Mode::Auto, state: ItemState::Running, diff --git a/src/engine/engine/mod.rs b/src/engine/engine/mod.rs index f8011b9..b9cc3f6 100644 --- a/src/engine/engine/mod.rs +++ b/src/engine/engine/mod.rs @@ -20,7 +20,7 @@ pub struct WatchList { #[derive(Clone)] pub struct Engine { pub terminal_server: Arc, - pub strategy: Arc, + pub strategy_engine: Arc, pub config: Arc>, pub accounts: Arc>, pub watch_list: Arc>, @@ -38,7 +38,7 @@ impl Engine { Ok(Arc::new_cyclic(|engine| Self { terminal_server: TerminalServer::new(engine.clone()), - strategy: strategy.initialize(engine.clone()), + strategy_engine: strategy.initialize(engine.clone()), config, accounts, watch_list: Arc::new(Mutex::new(WatchList { diff --git a/src/engine/engine/strategy.rs b/src/engine/engine/strategy.rs index d45db97..6c1d372 100644 --- a/src/engine/engine/strategy.rs +++ b/src/engine/engine/strategy.rs @@ -1,26 +1,24 @@ +use anyhow::Context; use hypersdk::hypercore::{self, CandleInterval, Subscription, WebSocket}; use pulse_sdk::prelude::*; use std::{ collections::HashSet, - path::PathBuf, - process::Stdio, sync::{Arc, Weak}, time::{SystemTime, UNIX_EPOCH}, }; -use tokio::{ - fs, - process::{Child, Command}, - sync::Mutex, -}; +use tokio::{fs, sync::Mutex}; use crate::{ engine::Engine, - store::{pulse_strategy, strategy::StrategyChild}, + store::{ + pulse_strategy, + strategy::{StrategyChild, get_manifest}, + }, }; pub struct StrategyEngine { pub engine: Weak, - pub child: Arc, + pub strategy: Mutex, pub ws: WebSocket, pub subscriptions: Mutex>, } @@ -42,7 +40,7 @@ impl StrategyEngine { )?; Ok(Self { - child: Arc::new(StrategyChild::new(strategy, strategy_manifest)), + strategy: Mutex::new(StrategyChild::new(strategy, strategy_manifest)), engine: Weak::new(), ws: hypercore::mainnet_ws(), subscriptions: Mutex::new(HashSet::new()), @@ -55,24 +53,66 @@ impl StrategyEngine { Arc::new(self) } - pub async fn run(&self) -> anyhow::Result<()> { + pub async fn send(self: &Arc, msg: &StrategyEngineMessage) -> tokio::io::Result<()> { + self.strategy.lock().await.send(msg).await + } + + pub async fn spawn(self: &Arc) { + let engine = self.clone(); + + tokio::spawn(async move { engine.run().await }); + } + + pub async fn reload(self: &Arc, id: &str) -> tokio::io::Result<()> { + { + let strategy = pulse_strategy(id)?; + + let (child, manifest) = get_manifest( + &strategy, + &strategy.join("strategy.bash"), + &fs::read(strategy.join("strategy.toml")).await?, + )?; + + let mut strategy = self.strategy.lock().await; + + strategy.child.kill().await?; + strategy.child = child; + strategy.manifest = manifest; + } + + let engine = self.clone(); + tokio::spawn(async move { engine.run().await }); + + Ok(()) + } + + pub async fn run(self: &Arc) -> anyhow::Result<()> { let engine = self .engine .upgrade() .expect("Failed to upgrade engine (StrategyEngine)"); - self.child.send(&StrategyEngineMessage::Initialize).await?; + let mut stdout = { + let mut child = self.strategy.lock().await; + + child + .child + .stdout + .take() + .context("Failed to obtain strategy stdout")? + }; + + self.send(&StrategyEngineMessage::Initialize).await?; loop { - match self.child.recv().await? { + match StrategyChild::read(&mut stdout).await? { None => {} Some(StrategyMessage::GetWatchList) => { - self.child - .send(&StrategyEngineMessage::WatchList( - engine.watch_list.lock().await.clone().items, - )) - .await?; + self.send(&StrategyEngineMessage::WatchList( + engine.watch_list.lock().await.clone().items, + )) + .await?; } Some(StrategyMessage::Log(mut log)) => { @@ -150,58 +190,16 @@ impl StrategyEngine { let start_time = now.saturating_sub(interval_ms * count as u64); - self.child - .send(&StrategyEngineMessage::Candlestick { - candles: client - .candle_snapshot(&symbol, interval, start_time, now) - .await?, - symbol, - interval, - }) - .await?; + self.send(&StrategyEngineMessage::Candlestick { + candles: client + .candle_snapshot(&symbol, interval, start_time, now) + .await?, + symbol, + interval, + }) + .await?; } } } } - - pub async fn spawn(self: &Arc) { - let engine = self.clone(); - - tokio::spawn(async move { engine.run().await }); - } - - pub async fn reload(self: &Arc, id: &str) -> tokio::io::Result<()> { - let strategy = pulse_strategy(id)?; - - let (child, manifest) = get_manifest( - &strategy, - &strategy.join("strategy.bash"), - &fs::read(strategy.join("strategy.toml")).await?, - )?; - - self.child.reload(child, manifest).await?; - - let engine = self.clone(); - tokio::spawn(async move { engine.run().await }); - - Ok(()) - } -} - -fn get_manifest<'de, M: serde::Deserialize<'de>>( - strategy_dir: &PathBuf, - strategy_path: &PathBuf, - manifest: &'de [u8], -) -> tokio::io::Result<(Child, M)> { - Ok(( - Command::new("bash") - .arg(strategy_path) - .current_dir(strategy_dir) - .stdin(Stdio::piped()) - .stdout(Stdio::piped()) - .stderr(Stdio::inherit()) - .spawn()?, - toml::from_slice(manifest) - .map_err(|v| tokio::io::Error::new(std::io::ErrorKind::InvalidInput, v.to_string()))?, - )) } diff --git a/src/engine/main.rs b/src/engine/main.rs index 1d45b2d..1393e79 100644 --- a/src/engine/main.rs +++ b/src/engine/main.rs @@ -9,7 +9,7 @@ async fn main() -> anyhow::Result<()> { let broadcaster = engine.spawn_broadcaster().await; - engine.strategy.spawn().await; + engine.strategy_engine.spawn().await; engine.terminal_server.run().await?; diff --git a/src/engine/store/strategy.rs b/src/engine/store/strategy.rs index fa7a99f..858fd9a 100644 --- a/src/engine/store/strategy.rs +++ b/src/engine/store/strategy.rs @@ -1,31 +1,25 @@ +use std::{path::PathBuf, process::Stdio}; + use pulse_sdk::{ map_postcard_err, strategy::{StrategyEngineMessage, StrategyManifest, StrategyMessage}, }; use tokio::{ io::{AsyncReadExt, AsyncWriteExt}, - process::{Child, ChildStdout}, - sync::Mutex, + process::{Child, ChildStdout, Command}, }; #[derive(Debug)] pub struct StrategyChild { - pub manifest: Mutex, - pub stdout: Mutex, - pub process: Mutex, + pub manifest: StrategyManifest, + pub child: Child, } impl StrategyChild { - pub fn new(mut child: Child, manifest: StrategyManifest) -> Self { - Self { - stdout: Mutex::new(child.stdout.take().expect("Failed to obtain child stdout")), - process: Mutex::new(child), - manifest: Mutex::new(manifest), - } + pub fn new(child: Child, manifest: StrategyManifest) -> Self { + Self { child, manifest } } - pub async fn recv(&self) -> tokio::io::Result> { - let mut stdout = self.stdout.lock().await; - + pub async fn read(stdout: &mut ChildStdout) -> tokio::io::Result> { let mut len_buf = [0u8; size_of::()]; let size = stdout.read_exact(&mut len_buf).await?; @@ -42,14 +36,13 @@ impl StrategyChild { Ok(Some(map_postcard_err(postcard::from_bytes(&buffer))?)) } - pub async fn send(&self, msg: &StrategyEngineMessage) -> tokio::io::Result<()> { + pub async fn send(&mut self, msg: &StrategyEngineMessage) -> tokio::io::Result<()> { self.send_raw(&map_postcard_err(postcard::to_allocvec(msg))?) .await } - pub async fn send_raw(&self, msg: &[u8]) -> tokio::io::Result<()> { - let mut process = self.process.lock().await; - let stdin = process.stdin.as_mut().unwrap(); + pub async fn send_raw(&mut self, msg: &[u8]) -> tokio::io::Result<()> { + let stdin = self.child.stdin.as_mut().unwrap(); stdin.write_all(&msg.len().to_le_bytes()).await?; stdin.write_all(msg).await?; @@ -57,20 +50,22 @@ impl StrategyChild { Ok(()) } +} - pub async fn reload( - &self, - mut child: Child, - manifest: StrategyManifest, - ) -> tokio::io::Result<()> { - let mut process = self.process.lock().await; - - process.kill().await?; - - *self.stdout.lock().await = child.stdout.take().expect("Failed to obtain child stdout"); - *self.manifest.lock().await = manifest; - *process = child; - - Ok(()) - } +pub fn get_manifest<'de, M: serde::Deserialize<'de>>( + strategy_dir: &PathBuf, + strategy_path: &PathBuf, + manifest: &'de [u8], +) -> tokio::io::Result<(Child, M)> { + Ok(( + Command::new("bash") + .arg(strategy_path) + .current_dir(strategy_dir) + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::inherit()) + .spawn()?, + toml::from_slice(manifest) + .map_err(|v| tokio::io::Error::new(std::io::ErrorKind::InvalidInput, v.to_string()))?, + )) } diff --git a/src/engine/terminal.rs b/src/engine/terminal.rs index a08f521..5dcfe20 100644 --- a/src/engine/terminal.rs +++ b/src/engine/terminal.rs @@ -65,7 +65,13 @@ impl TerminalServer { self.send_to( id, pulse_sdk::terminal::TerminalServerMessage::StrategyUpdated(Strategy { - strategy: engine.strategy.child.manifest.lock().await.clone(), + strategy: engine + .strategy_engine + .strategy + .lock() + .await + .manifest + .clone(), mode: Mode::Auto, state: ItemState::Running, cooldown: engine.config.lock().await.cooldown, From de0750ff2e1983ee33ee6b7e3f5d82e197f099ab Mon Sep 17 00:00:00 2001 From: Klesti Selimaj Date: Thu, 30 Jul 2026 16:52:35 +0200 Subject: [PATCH 05/12] Formatting --- src/terminal/formatting.rs | 28 ++++++++++++++++++++-------- 1 file changed, 20 insertions(+), 8 deletions(-) diff --git a/src/terminal/formatting.rs b/src/terminal/formatting.rs index ba695f4..80e7847 100644 --- a/src/terminal/formatting.rs +++ b/src/terminal/formatting.rs @@ -53,8 +53,8 @@ impl Formatted for SignalStatus { } else { format!("\x1b[31mSELL\x1b[0m") }, - format!("\x1b[35m{}\x1b[0m", signal.symbol), - signal.price.to_string(), + format_symbol(&signal.symbol), + format_usd(signal.price.as_f64()), format!("\x1b[33mAPR {}\x1b[0m", signal.confidence), ] } @@ -67,8 +67,8 @@ impl Formatted for SignalStatus { } else { format!("\x1b[31mSELL\x1b[0m") }, - format!("\x1b[35m{}\x1b[0m", signal.symbol), - signal.price.to_string(), + format_symbol(&signal.symbol), + format_usd(signal.price.as_f64()), format!("\x1b[33mAPR {}\x1b[0m", signal.confidence), ] } @@ -79,9 +79,9 @@ impl Formatted for SignalStatus { impl Formatted for MarketItem { fn get_formatted(&self) -> Vec { vec![ - self.symbol.to_string(), - self.price.to_string(), - self.volume_24h.to_string(), + format_symbol(&self.symbol), + format_usd(self.price.as_f64()), + format_usd(self.volume_24h.as_f64()), format!( "{} {}%", if self.trend.is_sign_positive() { @@ -98,7 +98,7 @@ impl Formatted for MarketItem { impl Formatted for Position { fn get_formatted(&self) -> Vec { vec![ - self.symbol.to_string(), + format_symbol(&self.symbol), format_f64(self.size.as_f64()), self.entry_price.to_string(), format!( @@ -232,6 +232,18 @@ pub fn apply_padding(mut items: Vec) -> Vec { items } +pub fn format_symbol(value: &str) -> String { + format!("\x1b[35m{value}\x1b[0m") +} + +pub fn format_usd(value: f64) -> String { + if value.is_sign_positive() { + format!("\x1b[32m${}\x1b[0m", format_f64(value)) + } else { + format!("\x1b[31m${}\x1b[0m", format_f64(value)) + } +} + pub fn format_f64(value: f64) -> String { let abs = value.abs(); From fb4d22c5a29f4518207ecd48d3b1d4023eab89ed Mon Sep 17 00:00:00 2001 From: Klesti Selimaj Date: Thu, 30 Jul 2026 23:56:31 +0200 Subject: [PATCH 06/12] repost tokio in sdk --- pulse-sdk/src/lib.rs | 1 + 1 file changed, 1 insertion(+) diff --git a/pulse-sdk/src/lib.rs b/pulse-sdk/src/lib.rs index eddadf3..afca266 100644 --- a/pulse-sdk/src/lib.rs +++ b/pulse-sdk/src/lib.rs @@ -15,6 +15,7 @@ pub mod prelude { pub use hypersdk; pub use postcard; + pub use tokio; } pub fn server_path() -> std::path::PathBuf { From ccac73c3c51364c6df08422d4cb6e85a78875a3e Mon Sep 17 00:00:00 2001 From: Klesti Selimaj Date: Fri, 31 Jul 2026 00:01:41 +0200 Subject: [PATCH 07/12] Strategy --- pulse-sdk/src/lib.rs | 12 +++++++++++- pulse-sdk/src/terminal.rs | 4 ++-- src/engine/engine/command.rs | 4 ++-- src/engine/terminal.rs | 2 +- src/terminal/formatting.rs | 2 +- src/terminal/main.rs | 2 +- src/terminal/terminal.rs | 2 +- 7 files changed, 19 insertions(+), 9 deletions(-) diff --git a/pulse-sdk/src/lib.rs b/pulse-sdk/src/lib.rs index afca266..ee5fe3e 100644 --- a/pulse-sdk/src/lib.rs +++ b/pulse-sdk/src/lib.rs @@ -5,13 +5,14 @@ pub use hypersdk; use tokio::io::{AsyncReadExt, AsyncWriteExt}; -use crate::strategy::StrategyEngineMessage; +use crate::{general::LogKind, strategy::StrategyEngineMessage}; pub mod prelude { pub use crate::general::*; pub use crate::server_path; pub use crate::strategy::*; pub use crate::terminal::*; + pub use crate::Strategy; pub use hypersdk; pub use postcard; @@ -63,6 +64,15 @@ pub trait Strategy { send_raw(&map_postcard_err(prelude::postcard::to_allocvec(msg))?).await } + async fn log(&self, kind: LogKind, name: &str, message: &str) -> tokio::io::Result<()> { + self.send(&prelude::StrategyMessage::Log(prelude::EventLog { + kind, + name: name.to_owned(), + message: message.to_owned(), + })) + .await + } + async fn on_raw(&self, data: &[u8]) -> tokio::io::Result<()> { match map_postcard_err(postcard::from_bytes(&data))? { StrategyEngineMessage::Initialize => self.initialize().await, diff --git a/pulse-sdk/src/terminal.rs b/pulse-sdk/src/terminal.rs index 67e5ac6..6d52182 100644 --- a/pulse-sdk/src/terminal.rs +++ b/pulse-sdk/src/terminal.rs @@ -15,7 +15,7 @@ pub enum TerminalServerMessage { PositionsUpdated(Vec), // Strategy - StrategyUpdated(Strategy), + StrategyUpdated(StrategyStatus), // Signals SignalsUpdated(Vec), @@ -80,7 +80,7 @@ pub enum ItemState { } #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] -pub struct Strategy { +pub struct StrategyStatus { pub strategy: StrategyManifest, pub mode: Mode, diff --git a/src/engine/engine/command.rs b/src/engine/engine/command.rs index 91c2c01..577ab31 100644 --- a/src/engine/engine/command.rs +++ b/src/engine/engine/command.rs @@ -1,6 +1,6 @@ use crate::{engine::Engine, store::config::Config}; -use pulse_sdk::terminal::{ItemState, Mode, Strategy}; +use pulse_sdk::terminal::{ItemState, Mode, StrategyStatus}; use toml::Value; impl Engine { @@ -74,7 +74,7 @@ impl Engine { self.config.lock().await.cooldown = cooldown; self.terminal_server.broadcast( - pulse_sdk::terminal::TerminalServerMessage::StrategyUpdated(Strategy { + pulse_sdk::terminal::TerminalServerMessage::StrategyUpdated(StrategyStatus { strategy: self.strategy_engine.strategy.lock().await.manifest.clone(), mode: Mode::Auto, diff --git a/src/engine/terminal.rs b/src/engine/terminal.rs index 5dcfe20..619d564 100644 --- a/src/engine/terminal.rs +++ b/src/engine/terminal.rs @@ -64,7 +64,7 @@ impl TerminalServer { self.send_to( id, - pulse_sdk::terminal::TerminalServerMessage::StrategyUpdated(Strategy { + pulse_sdk::terminal::TerminalServerMessage::StrategyUpdated(StrategyStatus { strategy: engine .strategy_engine .strategy diff --git a/src/terminal/formatting.rs b/src/terminal/formatting.rs index 80e7847..4e5c3db 100644 --- a/src/terminal/formatting.rs +++ b/src/terminal/formatting.rs @@ -206,7 +206,7 @@ impl Formatted for Vec { } } -impl Formatted for Strategy { +impl Formatted for StrategyStatus { fn get_formatted(&self) -> Vec { vec![ Triple( diff --git a/src/terminal/main.rs b/src/terminal/main.rs index e751356..070d064 100644 --- a/src/terminal/main.rs +++ b/src/terminal/main.rs @@ -29,7 +29,7 @@ pub struct PulseTradeApp { command: State, scroll: State>, - strategy: State>, + strategy: State>, status: State>, watch_list: State>, diff --git a/src/terminal/terminal.rs b/src/terminal/terminal.rs index 608d1b8..ccc646d 100644 --- a/src/terminal/terminal.rs +++ b/src/terminal/terminal.rs @@ -82,7 +82,7 @@ impl TerminalClient { active_positions: State>, logs: State>, signals: State>, - market_overview: State>, + market_overview: State>, status: State>, inspect: State, ) -> tokio::io::Result<()> { From e4ba12c5de6894063e937d897ef91afc582574a7 Mon Sep 17 00:00:00 2001 From: Klesti Selimaj Date: Fri, 31 Jul 2026 00:56:16 +0200 Subject: [PATCH 08/12] Strategy command --- pulse-sdk/src/lib.rs | 3 +-- src/engine/engine/command.rs | 23 ++++++++++++++++++++--- 2 files changed, 21 insertions(+), 5 deletions(-) diff --git a/pulse-sdk/src/lib.rs b/pulse-sdk/src/lib.rs index ee5fe3e..af0744e 100644 --- a/pulse-sdk/src/lib.rs +++ b/pulse-sdk/src/lib.rs @@ -8,15 +8,14 @@ use tokio::io::{AsyncReadExt, AsyncWriteExt}; use crate::{general::LogKind, strategy::StrategyEngineMessage}; pub mod prelude { + pub use crate::Strategy; pub use crate::general::*; pub use crate::server_path; pub use crate::strategy::*; pub use crate::terminal::*; - pub use crate::Strategy; pub use hypersdk; pub use postcard; - pub use tokio; } pub fn server_path() -> std::path::PathBuf { diff --git a/src/engine/engine/command.rs b/src/engine/engine/command.rs index 577ab31..5cee75e 100644 --- a/src/engine/engine/command.rs +++ b/src/engine/engine/command.rs @@ -1,10 +1,14 @@ use crate::{engine::Engine, store::config::Config}; -use pulse_sdk::terminal::{ItemState, Mode, StrategyStatus}; +use pulse_sdk::prelude::*; use toml::Value; impl Engine { - pub async fn execute_command(&self, command: &str, args: Vec<&str>) -> tokio::io::Result<()> { + pub async fn execute_command( + &self, + command: &str, + mut args: Vec<&str>, + ) -> tokio::io::Result<()> { match command { "config" | "cfg" => { if args.len() == 0 { @@ -43,7 +47,7 @@ impl Engine { .await?; } - "strategy" | "strat" | "str" | "sg" => { + "strategy" | "strat" | "sg" => { set_cfg!(id, { let id: String = id; @@ -166,6 +170,19 @@ impl Engine { } } + "strategy" | "strat" | "sg" => { + let Some(strategy_command) = args.drain(0..=0).next() else { + return self.invalid_command_usage("strategy").await; + }; + + self.strategy_engine + .send(&StrategyEngineMessage::Command { + command: strategy_command.to_owned(), + args: args.into_iter().map(Into::into).collect(), + }) + .await?; + } + _ => { self.terminal_server .error( From fccd03c3dfee551de118c9d3cb5b7b679f7a2987 Mon Sep 17 00:00:00 2001 From: Klesti Selimaj Date: Fri, 31 Jul 2026 01:17:35 +0200 Subject: [PATCH 09/12] Strategy split ws --- Cargo.lock | 29 +++++++++++++++-------------- Cargo.toml | 1 + src/engine/engine/mod.rs | 14 +++++++++++--- src/engine/engine/strategy.rs | 14 +++++++------- 4 files changed, 34 insertions(+), 24 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 2242481..36a13bf 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2432,9 +2432,9 @@ checksum = "e6d5a32815ae3f33302d95fdcb2ce17862f8c65363dcfd29360480ba1001fc9c" [[package]] name = "futures" -version = "0.3.32" +version = "0.3.33" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8b147ee9d1f6d097cef9ce628cd2ee62288d963e16fb287bd9286455b241382d" +checksum = "a88cf1f829d945f548cf8fec32c61b1f202b6d93b45848602fc02af4b12ad218" dependencies = [ "futures-channel", "futures-core", @@ -2447,9 +2447,9 @@ dependencies = [ [[package]] name = "futures-channel" -version = "0.3.32" +version = "0.3.33" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "07bbe89c50d7a535e539b8c17bc0b49bdb77747034daa8087407d655f3f7cc1d" +checksum = "262590f4fe6afeb0bc83be1daa64e52657fe185690a958af7f3ad0e92085c5ae" dependencies = [ "futures-core", "futures-sink", @@ -2457,15 +2457,15 @@ dependencies = [ [[package]] name = "futures-core" -version = "0.3.32" +version = "0.3.33" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7e3450815272ef58cec6d564423f6e755e25379b217b0bc688e295ba24df6b1d" +checksum = "2cd50c473c80f6d7c3670a752354b8e569b1a7cbfdc0419ec88e5edad85e0dc7" [[package]] name = "futures-executor" -version = "0.3.32" +version = "0.3.33" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "baf29c38818342a3b26b5b923639e7b1f4a61fc5e76102d4b1981c6dc7a7579d" +checksum = "6754879cc9f2c66f88c6e5c35344bb0bdb0708b0352b1201815667c7eabc7458" dependencies = [ "futures-core", "futures-task", @@ -2480,9 +2480,9 @@ checksum = "4577ecaa3c4f96589d473f679a71b596316f6641bc350038b962a5daf0085d7a" [[package]] name = "futures-macro" -version = "0.3.32" +version = "0.3.33" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e835b70203e41293343137df5c0664546da5745f82ec9b84d40be8336958447b" +checksum = "2d6d3cde68c518367be28956066ddfef33813991b77a55005a69dae04bf3b10b" dependencies = [ "proc-macro2", "quote", @@ -2497,15 +2497,15 @@ checksum = "e34418ac499d6305c2fb5ad0ed2f6ac998c5f8ca209b4510f7f94242c647e307" [[package]] name = "futures-task" -version = "0.3.32" +version = "0.3.33" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "037711b3d59c33004d3856fbdc83b99d4ff37a24768fa1be9ce3538a1cde4393" +checksum = "b231ed28831efb4a61a08580c4bc233ec56bc009f4cd8f52da2c3cb97df0c109" [[package]] name = "futures-util" -version = "0.3.32" +version = "0.3.33" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "389ca41296e6190b48053de0321d02a77f32f8a5d2461dd38762c0593805c6d6" +checksum = "a77a90a256fce34da66415271e30f94ee91c57b04b8a2c042d9cf3220179deaa" dependencies = [ "futures-channel", "futures-core", @@ -3959,6 +3959,7 @@ dependencies = [ "anyhow", "chrono", "crossterm", + "futures", "hypersdk", "postcard", "pulse-sdk", diff --git a/Cargo.toml b/Cargo.toml index c1d3ab1..2209230 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -25,6 +25,7 @@ chrono = "0.4.45" serde_json = "1" rand = "0.8.7" toml = "1.1.3" +futures = "0.3.33" [workspace] members = ["pulse-ui", "pulse-sdk"] diff --git a/src/engine/engine/mod.rs b/src/engine/engine/mod.rs index b9cc3f6..89bc936 100644 --- a/src/engine/engine/mod.rs +++ b/src/engine/engine/mod.rs @@ -7,6 +7,7 @@ use crate::{ store::{accounts::AccountList, config::Config}, terminal::TerminalServer, }; +use hypersdk::hypercore::ws::ConnectionStream; use pulse_sdk::prelude::*; use std::{collections::HashMap, sync::Arc}; use tokio::{sync::Mutex, task::JoinHandle}; @@ -19,8 +20,12 @@ pub struct WatchList { #[derive(Clone)] pub struct Engine { + // engine pub terminal_server: Arc, pub strategy_engine: Arc, + pub ws_stream: Arc>, + + // data pub config: Arc>, pub accounts: Arc>, pub watch_list: Arc>, @@ -31,16 +36,19 @@ impl Engine { pub async fn new() -> tokio::io::Result> { let config = Config::new().await?; - let strategy = StrategyEngine::new(&config.strategy).await?; + let (ws_handle, ws_stream) = hypersdk::hypercore::mainnet_ws().split(); + + let strategy = StrategyEngine::new(&config.strategy, ws_handle).await?; let accounts = Arc::new(Mutex::new(AccountList::new().await?)); let config = Arc::new(Mutex::new(config)); Ok(Arc::new_cyclic(|engine| Self { - terminal_server: TerminalServer::new(engine.clone()), - strategy_engine: strategy.initialize(engine.clone()), config, accounts, + ws_stream: Arc::new(Mutex::new(ws_stream)), + terminal_server: TerminalServer::new(engine.clone()), + strategy_engine: strategy.initialize(engine.clone()), watch_list: Arc::new(Mutex::new(WatchList { name_to_index: HashMap::new(), items: Vec::new(), diff --git a/src/engine/engine/strategy.rs b/src/engine/engine/strategy.rs index 6c1d372..90da13c 100644 --- a/src/engine/engine/strategy.rs +++ b/src/engine/engine/strategy.rs @@ -1,5 +1,5 @@ use anyhow::Context; -use hypersdk::hypercore::{self, CandleInterval, Subscription, WebSocket}; +use hypersdk::hypercore::{self, CandleInterval, Subscription, ws::ConnectionHandle}; use pulse_sdk::prelude::*; use std::{ collections::HashSet, @@ -19,12 +19,12 @@ use crate::{ pub struct StrategyEngine { pub engine: Weak, pub strategy: Mutex, - pub ws: WebSocket, + pub ws_handle: ConnectionHandle, pub subscriptions: Mutex>, } impl StrategyEngine { - pub async fn new(strategy_id: &str) -> tokio::io::Result { + pub async fn new(strategy_id: &str, ws_handle: ConnectionHandle) -> tokio::io::Result { let strategy = pulse_strategy(strategy_id)?; let (strategy, strategy_manifest) = get_manifest( @@ -42,7 +42,7 @@ impl StrategyEngine { Ok(Self { strategy: Mutex::new(StrategyChild::new(strategy, strategy_manifest)), engine: Weak::new(), - ws: hypercore::mainnet_ws(), + ws_handle, subscriptions: Mutex::new(HashSet::new()), }) } @@ -142,18 +142,18 @@ impl StrategyEngine { } Some(StrategyMessage::Subscribe(subscription)) => { - self.ws.subscribe(subscription.clone()); + self.ws_handle.subscribe(subscription.clone()); self.subscriptions.lock().await.insert(subscription); } Some(StrategyMessage::Unsubscribe(subscription)) => { self.subscriptions.lock().await.remove(&subscription); - self.ws.unsubscribe(subscription); + self.ws_handle.unsubscribe(subscription); } Some(StrategyMessage::UnsubscribeAll) => { for sub in self.subscriptions.lock().await.drain() { - self.ws.unsubscribe(sub); + self.ws_handle.unsubscribe(sub); } } From b5a1a07cf013b8494093d69fb73a91733a199915 Mon Sep 17 00:00:00 2001 From: Klesti Selimaj Date: Fri, 31 Jul 2026 01:26:03 +0200 Subject: [PATCH 10/12] Refactored engine --- src/engine/engine/command.rs | 4 ++ src/engine/engine/mod.rs | 85 +----------------------------------- src/engine/main.rs | 6 +-- src/engine/terminal.rs | 84 ++++++++++++++++++++++++++++++++++- 4 files changed, 91 insertions(+), 88 deletions(-) diff --git a/src/engine/engine/command.rs b/src/engine/engine/command.rs index 5cee75e..c3c7a75 100644 --- a/src/engine/engine/command.rs +++ b/src/engine/engine/command.rs @@ -4,6 +4,10 @@ use pulse_sdk::prelude::*; use toml::Value; impl Engine { + pub async fn invalid_command_usage(&self, name: &str) -> tokio::io::Result<()> { + self.terminal_server.error(name, "Invalid usage").await + } + pub async fn execute_command( &self, command: &str, diff --git a/src/engine/engine/mod.rs b/src/engine/engine/mod.rs index 89bc936..d0d6f5c 100644 --- a/src/engine/engine/mod.rs +++ b/src/engine/engine/mod.rs @@ -10,7 +10,7 @@ use crate::{ use hypersdk::hypercore::ws::ConnectionStream; use pulse_sdk::prelude::*; use std::{collections::HashMap, sync::Arc}; -use tokio::{sync::Mutex, task::JoinHandle}; +use tokio::sync::Mutex; #[derive(Debug, Clone)] pub struct WatchList { @@ -56,87 +56,4 @@ impl Engine { signals: Arc::new(Mutex::new(Vec::new())), })) } - - pub async fn spawn_broadcaster(&self) -> JoinHandle> { - let s = self.clone(); - - tokio::spawn(async move { s.run_broadcaster().await }) - } - - pub async fn run_broadcaster(&self) -> tokio::io::Result<()> { - let mut refresh = tokio::time::interval(tokio::time::Duration::from_secs(5)); - - let client = hypersdk::hypercore::mainnet(); - - loop { - refresh.tick().await; - - let watch_list = &self.config.lock().await.watchlist; - - match crate::fetch::fetch_watch_list(&client, watch_list).await { - Ok(watch_list) => { - *self.watch_list.lock().await = watch_list.clone(); - - if let Err(error) = self - .terminal_server - .broadcast(TerminalServerMessage::WatchListUpdated(watch_list.items)) - .await - { - self.terminal_server - .error( - "Broadcaster", - &format!("Failed to broadcast HyperLiquid watch list: {error}"), - ) - .await? - } - } - - Err(error) => { - self.terminal_server - .error( - "Broadcaster", - &format!("Failed to refresh HyperLiquid watch list: {error}"), - ) - .await? - } - } - - let accounts = self.accounts.lock().await; - - if let Some(acc) = accounts.get_active() { - match client.clearinghouse_state(acc.address, None).await { - Ok(state) => { - self.terminal_server - .broadcast(TerminalServerMessage::PositionsUpdated( - state - .asset_positions - .into_iter() - .map(|position| Position { - symbol: position.position.coin, - size: position.position.szi, - entry_price: position.position.entry_px.unwrap_or_default(), - pnl: position.position.unrealized_pnl, - }) - .collect(), - )) - .await?; - } - Err(e) => { - self.terminal_server - .error("orders", &format!("Unable to get open orders: {e}")) - .await?; - } - } - } else { - self.terminal_server.error( - "orders", - "Unable to get active account, make sure you have configured accounts properly", - ).await?; - } - } - } - - pub async fn invalid_command_usage(&self, name: &str) -> tokio::io::Result<()> { - self.terminal_server.error(name, "Invalid usage").await - } } diff --git a/src/engine/main.rs b/src/engine/main.rs index 1393e79..af859e4 100644 --- a/src/engine/main.rs +++ b/src/engine/main.rs @@ -7,12 +7,12 @@ pub mod terminal; async fn main() -> anyhow::Result<()> { let engine = engine::Engine::new().await?; - let broadcaster = engine.spawn_broadcaster().await; + let server = engine.terminal_server.spawn_server().await; + let broadcaster = engine.terminal_server.spawn_broadcaster().await; engine.strategy_engine.spawn().await; - engine.terminal_server.run().await?; - + server.await??; broadcaster.await??; Ok(()) diff --git a/src/engine/terminal.rs b/src/engine/terminal.rs index 619d564..e308812 100644 --- a/src/engine/terminal.rs +++ b/src/engine/terminal.rs @@ -11,6 +11,7 @@ use tokio::{ unix::{OwnedReadHalf, OwnedWriteHalf}, }, sync::Mutex, + task::JoinHandle, }; #[derive(Debug)] @@ -29,7 +30,13 @@ impl TerminalServer { }) } - pub async fn run(self: &Arc) -> tokio::io::Result<()> { + pub async fn spawn_server(self: &Arc) -> JoinHandle> { + let s = self.clone(); + + tokio::spawn(s.run_server()) + } + + pub async fn run_server(self: Arc) -> tokio::io::Result<()> { let path = pulse_sdk::server_path(); if path.exists() { @@ -151,6 +158,81 @@ impl TerminalServer { Ok(()) } + pub async fn spawn_broadcaster(self: &Arc) -> JoinHandle> { + let s = self.clone(); + + tokio::spawn(s.run_broadcaster()) + } + + pub async fn run_broadcaster(self: Arc) -> tokio::io::Result<()> { + let engine = self.get_engine(); + + let mut refresh = tokio::time::interval(tokio::time::Duration::from_secs(5)); + + let client = hypersdk::hypercore::mainnet(); + + loop { + refresh.tick().await; + + match crate::fetch::fetch_watch_list(&client, &engine.config.lock().await.watchlist) + .await + { + Ok(watch_list) => { + *engine.watch_list.lock().await = watch_list.clone(); + + if let Err(error) = self + .broadcast(TerminalServerMessage::WatchListUpdated(watch_list.items)) + .await + { + self.error( + "Broadcaster", + &format!("Failed to broadcast HyperLiquid watch list: {error}"), + ) + .await? + } + } + + Err(error) => { + self.error( + "Broadcaster", + &format!("Failed to refresh HyperLiquid watch list: {error}"), + ) + .await? + } + } + + if let Some(acc) = engine.accounts.lock().await.get_active() { + match client.clearinghouse_state(acc.address, None).await { + Ok(state) => { + self.broadcast(TerminalServerMessage::PositionsUpdated( + state + .asset_positions + .into_iter() + .map(|position| Position { + symbol: position.position.coin, + size: position.position.szi, + entry_price: position.position.entry_px.unwrap_or_default(), + pnl: position.position.unrealized_pnl, + }) + .collect(), + )) + .await?; + } + Err(e) => { + self.error("orders", &format!("Unable to get open orders: {e}")) + .await?; + } + } + } else { + self.error( + "orders", + "Unable to get active account, make sure you have configured accounts properly", + ) + .await?; + } + } + } + pub async fn send_to( self: &Arc, id: &usize, From f3e2deab09e33acdb0249e6139d779b1fde5837a Mon Sep 17 00:00:00 2001 From: Klesti Selimaj Date: Fri, 31 Jul 2026 01:42:45 +0200 Subject: [PATCH 11/12] Refactored engine --- .DS_Store | Bin 0 -> 6148 bytes src/engine/engine/mod.rs | 16 ++++++++++++-- src/engine/engine/strategy.rs | 31 +++++++++------------------- src/engine/{ => engine}/terminal.rs | 0 src/engine/main.rs | 3 +-- 5 files changed, 25 insertions(+), 25 deletions(-) create mode 100644 .DS_Store rename src/engine/{ => engine}/terminal.rs (100%) diff --git a/.DS_Store b/.DS_Store new file mode 100644 index 0000000000000000000000000000000000000000..fe578e82f53320d6bd0bf92699ea4c032cc3ed2f GIT binary patch literal 6148 zcmeHKJ4*vW5T4DQay|%GSXiDSb_!{1J;Mo>LK@4M2T_u6rzUoRE9?Xk?ib>bS%d1v>-N}>3; zVf!0n{mGv3ZD>DXDojBFedxduG+_n%2DB{(uo&e>nOw?Sr^)rKY^9@$D{ATUj!C=Bs&&ijV(Or2tc!0bZgNM7c zEU%lco$cooz&)GHn^DL;E5Hh{0-p-d_aVSW7-~!v%1;M&@(6(F;^e zh3G*ODiu+s3Vp;7Djnwo9T#d$6{>U)vNEotD+_%>5wbeY2QnN)sE~VBfEBP6aMVwq z_W#ZA&;NE3|6&DLfq$idaHC38#!J$BYv$r;uMMzOv5`?+s!*n2r?+D{psjcwn>6$V XG(ZeBrV4QejsFNp8MtEwepP`FQr)L| literal 0 HcmV?d00001 diff --git a/src/engine/engine/mod.rs b/src/engine/engine/mod.rs index d0d6f5c..045b85d 100644 --- a/src/engine/engine/mod.rs +++ b/src/engine/engine/mod.rs @@ -1,11 +1,11 @@ pub mod command; pub mod execution; pub mod strategy; +pub mod terminal; use crate::{ - engine::strategy::StrategyEngine, + engine::{strategy::StrategyEngine, terminal::TerminalServer}, store::{accounts::AccountList, config::Config}, - terminal::TerminalServer, }; use hypersdk::hypercore::ws::ConnectionStream; use pulse_sdk::prelude::*; @@ -56,4 +56,16 @@ impl Engine { signals: Arc::new(Mutex::new(Vec::new())), })) } + + /// Starts the main strategy server + /// When a strategy is reloaded it restarts automatically + pub async fn run(&self) -> anyhow::Result<()> { + loop { + self.strategy_engine.run().await?; + + self.terminal_server + .info("engine::main", "Strategy stopped, restarting") + .await?; + } + } } diff --git a/src/engine/engine/strategy.rs b/src/engine/engine/strategy.rs index 90da13c..b5fb0cf 100644 --- a/src/engine/engine/strategy.rs +++ b/src/engine/engine/strategy.rs @@ -57,31 +57,20 @@ impl StrategyEngine { self.strategy.lock().await.send(msg).await } - pub async fn spawn(self: &Arc) { - let engine = self.clone(); - - tokio::spawn(async move { engine.run().await }); - } - pub async fn reload(self: &Arc, id: &str) -> tokio::io::Result<()> { - { - let strategy = pulse_strategy(id)?; + let strategy = pulse_strategy(id)?; - let (child, manifest) = get_manifest( - &strategy, - &strategy.join("strategy.bash"), - &fs::read(strategy.join("strategy.toml")).await?, - )?; + let (child, manifest) = get_manifest( + &strategy, + &strategy.join("strategy.bash"), + &fs::read(strategy.join("strategy.toml")).await?, + )?; - let mut strategy = self.strategy.lock().await; + let mut strategy = self.strategy.lock().await; - strategy.child.kill().await?; - strategy.child = child; - strategy.manifest = manifest; - } - - let engine = self.clone(); - tokio::spawn(async move { engine.run().await }); + strategy.child.kill().await?; + strategy.child = child; + strategy.manifest = manifest; Ok(()) } diff --git a/src/engine/terminal.rs b/src/engine/engine/terminal.rs similarity index 100% rename from src/engine/terminal.rs rename to src/engine/engine/terminal.rs diff --git a/src/engine/main.rs b/src/engine/main.rs index af859e4..c9d3244 100644 --- a/src/engine/main.rs +++ b/src/engine/main.rs @@ -1,7 +1,6 @@ pub mod engine; pub mod fetch; pub mod store; -pub mod terminal; #[tokio::main] async fn main() -> anyhow::Result<()> { @@ -10,7 +9,7 @@ async fn main() -> anyhow::Result<()> { let server = engine.terminal_server.spawn_server().await; let broadcaster = engine.terminal_server.spawn_broadcaster().await; - engine.strategy_engine.spawn().await; + engine.run().await?; server.await??; broadcaster.await??; From c9a37e40245e4d7d39ecd85e90b00d5e18b26208 Mon Sep 17 00:00:00 2001 From: Klesti Selimaj Date: Fri, 31 Jul 2026 02:21:17 +0200 Subject: [PATCH 12/12] Refactored entry price --- pulse-sdk/src/general.rs | 2 +- src/engine/engine/execution.rs | 6 +++--- src/terminal/formatting.rs | 4 ++-- 3 files changed, 6 insertions(+), 6 deletions(-) diff --git a/pulse-sdk/src/general.rs b/pulse-sdk/src/general.rs index 1df15fd..5b5a21f 100644 --- a/pulse-sdk/src/general.rs +++ b/pulse-sdk/src/general.rs @@ -21,7 +21,7 @@ pub struct Signal { pub symbol: String, pub side: Side, pub confidence: f32, - pub price: Decimal, + pub entry_price: Decimal, pub take_profit: Decimal, pub stop_loss: Decimal, } diff --git a/src/engine/engine/execution.rs b/src/engine/engine/execution.rs index a5e1084..a7370c0 100644 --- a/src/engine/engine/execution.rs +++ b/src/engine/engine/execution.rs @@ -35,7 +35,7 @@ impl Engine { OrderRequest { asset: asset_id, is_buy: matches!(signal.side, Side::Bid), - limit_px: signal.price, + limit_px: signal.entry_price, sz: 0.into(), reduce_only: false, order_type: OrderTypePlacement::Limit { @@ -46,7 +46,7 @@ impl Engine { OrderRequest { asset: asset_id, is_buy: matches!(signal.side, Side::Bid), - limit_px: signal.price, + limit_px: signal.entry_price, sz: 0.into(), reduce_only: true, order_type: OrderTypePlacement::Trigger { @@ -59,7 +59,7 @@ impl Engine { OrderRequest { asset: asset_id, is_buy: matches!(signal.side, Side::Bid), - limit_px: signal.price, + limit_px: signal.entry_price, sz: 0.into(), reduce_only: true, order_type: OrderTypePlacement::Trigger { diff --git a/src/terminal/formatting.rs b/src/terminal/formatting.rs index 4e5c3db..1fd6fdb 100644 --- a/src/terminal/formatting.rs +++ b/src/terminal/formatting.rs @@ -54,7 +54,7 @@ impl Formatted for SignalStatus { format!("\x1b[31mSELL\x1b[0m") }, format_symbol(&signal.symbol), - format_usd(signal.price.as_f64()), + format_usd(signal.entry_price.as_f64()), format!("\x1b[33mAPR {}\x1b[0m", signal.confidence), ] } @@ -68,7 +68,7 @@ impl Formatted for SignalStatus { format!("\x1b[31mSELL\x1b[0m") }, format_symbol(&signal.symbol), - format_usd(signal.price.as_f64()), + format_usd(signal.entry_price.as_f64()), format!("\x1b[33mAPR {}\x1b[0m", signal.confidence), ] }