diff --git a/.DS_Store b/.DS_Store new file mode 100644 index 0000000..fe578e8 Binary files /dev/null and b/.DS_Store differ 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/pulse-sdk/src/general.rs b/pulse-sdk/src/general.rs index f36fc66..5b5a21f 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 entry_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..af0744e 100644 --- a/pulse-sdk/src/lib.rs +++ b/pulse-sdk/src/lib.rs @@ -1,19 +1,18 @@ pub mod general; pub mod strategy; pub mod terminal; -pub mod units; pub use hypersdk; use tokio::io::{AsyncReadExt, AsyncWriteExt}; -use crate::strategy::StrategyEngineMessage; +use crate::{general::LogKind, strategy::StrategyEngineMessage}; pub mod prelude { + pub use crate::Strategy; 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; @@ -37,41 +36,41 @@ 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 - } - }; -} - #[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 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))? { diff --git a/pulse-sdk/src/terminal.rs b/pulse-sdk/src/terminal.rs index efac397..6d52182 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}; @@ -16,7 +15,7 @@ pub enum TerminalServerMessage { PositionsUpdated(Vec), // Strategy - StrategyUpdated(Strategy), + StrategyUpdated(StrategyStatus), // Signals SignalsUpdated(Vec), @@ -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)] @@ -108,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, @@ -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/command.rs b/src/engine/engine/command.rs index 1facea8..c3c7a75 100644 --- a/src/engine/engine/command.rs +++ b/src/engine/engine/command.rs @@ -1,10 +1,18 @@ use crate::{engine::Engine, store::config::Config}; -use pulse_sdk::terminal::{ItemState, Mode, Strategy}; +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 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, + mut args: Vec<&str>, + ) -> tokio::io::Result<()> { match command { "config" | "cfg" => { if args.len() == 0 { @@ -43,7 +51,7 @@ impl Engine { .await?; } - "strategy" | "strat" | "str" | "sg" => { + "strategy" | "strat" | "sg" => { set_cfg!(id, { let id: String = id; @@ -60,7 +68,7 @@ impl Engine { .await; } - self.strategy.reload_strategy(id.as_str()).await?; + self.strategy_engine.reload(id.as_str()).await?; self.config.lock().await.strategy = id; }); @@ -74,8 +82,8 @@ impl Engine { self.config.lock().await.cooldown = cooldown; self.terminal_server.broadcast( - pulse_sdk::terminal::TerminalServerMessage::StrategyUpdated(Strategy { - strategy: self.strategy.strategy.manifest.lock().await.clone(), + pulse_sdk::terminal::TerminalServerMessage::StrategyUpdated(StrategyStatus { + strategy: self.strategy_engine.strategy.lock().await.manifest.clone(), mode: Mode::Auto, state: ItemState::Running, @@ -166,6 +174,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( diff --git a/src/engine/engine/execution.rs b/src/engine/engine/execution.rs index 9d56b4e..a7370c0 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.entry_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.entry_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.entry_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..045b85d 100644 --- a/src/engine/engine/mod.rs +++ b/src/engine/engine/mod.rs @@ -1,15 +1,16 @@ 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::*; use std::{collections::HashMap, sync::Arc}; -use tokio::{sync::Mutex, task::JoinHandle}; +use tokio::sync::Mutex; #[derive(Debug, Clone)] pub struct WatchList { @@ -19,8 +20,12 @@ pub struct WatchList { #[derive(Clone)] pub struct Engine { + // engine pub terminal_server: Arc, - pub strategy: 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: 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(), @@ -49,89 +57,15 @@ impl Engine { })) } - 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(); - + /// Starts the main strategy server + /// When a strategy is reloaded it restarts automatically + pub async fn run(&self) -> anyhow::Result<()> { loop { - refresh.tick().await; + self.strategy_engine.run().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: 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), - }) - .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?; - } + self.terminal_server + .info("engine::main", "Strategy stopped, restarting") + .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/engine/strategy.rs b/src/engine/engine/strategy.rs index 8f8f074..b5fb0cf 100644 --- a/src/engine/engine/strategy.rs +++ b/src/engine/engine/strategy.rs @@ -1,45 +1,48 @@ -use hypersdk::hypercore::{self, CandleInterval, Subscription, WebSocket}; +use anyhow::Context; +use hypersdk::hypercore::{self, CandleInterval, Subscription, ws::ConnectionHandle}; 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 strategy: Arc, pub engine: Weak, - - pub ws: WebSocket, + pub strategy: Mutex, + 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( &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 { - strategy: Arc::new(StrategyChild::new(strategy, strategy_manifest)), + strategy: Mutex::new(StrategyChild::new(strategy, strategy_manifest)), engine: Weak::new(), - ws: hypercore::mainnet_ws(), + ws_handle, subscriptions: Mutex::new(HashSet::new()), }) } @@ -50,26 +53,55 @@ impl StrategyEngine { Arc::new(self) } - pub async fn run_strategy(&self) -> anyhow::Result<()> { + pub async fn send(self: &Arc, msg: &StrategyEngineMessage) -> tokio::io::Result<()> { + self.strategy.lock().await.send(msg).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; + + Ok(()) + } + + pub async fn run(self: &Arc) -> anyhow::Result<()> { let engine = self .engine .upgrade() .expect("Failed to upgrade engine (StrategyEngine)"); - self.strategy - .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.strategy.recv().await? { + match StrategyChild::read(&mut stdout).await? { None => {} Some(StrategyMessage::GetWatchList) => { - self.strategy - .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)) => { @@ -99,18 +131,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); } } @@ -147,58 +179,16 @@ impl StrategyEngine { let start_time = now.saturating_sub(interval_ms * count as u64); - self.strategy - .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_strategy().await }); - } - - pub async fn reload_strategy(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.strategy.reload(child, manifest).await?; - - let engine = self.clone(); - tokio::spawn(async move { engine.run_strategy().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/terminal.rs b/src/engine/engine/terminal.rs similarity index 64% rename from src/engine/terminal.rs rename to src/engine/engine/terminal.rs index 97e8150..e308812 100644 --- a/src/engine/terminal.rs +++ b/src/engine/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() { @@ -64,8 +71,14 @@ impl TerminalServer { self.send_to( id, - pulse_sdk::terminal::TerminalServerMessage::StrategyUpdated(Strategy { - strategy: engine.strategy.strategy.manifest.lock().await.clone(), + pulse_sdk::terminal::TerminalServerMessage::StrategyUpdated(StrategyStatus { + strategy: engine + .strategy_engine + .strategy + .lock() + .await + .manifest + .clone(), mode: Mode::Auto, state: ItemState::Running, cooldown: engine.config.lock().await.cooldown, @@ -145,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, 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/engine/main.rs b/src/engine/main.rs index 1d45b2d..c9d3244 100644 --- a/src/engine/main.rs +++ b/src/engine/main.rs @@ -1,18 +1,17 @@ pub mod engine; pub mod fetch; pub mod store; -pub mod terminal; #[tokio::main] 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.spawn().await; - - engine.terminal_server.run().await?; + engine.run().await?; + server.await??; broadcaster.await??; Ok(()) 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/terminal/formatting.rs b/src/terminal/formatting.rs index dace9dd..1fd6fdb 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,13 +48,13 @@ 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") }, - format!("\x1b[35m{}\x1b[0m", signal.symbol), - signal.price.to_string(), + format_symbol(&signal.symbol), + format_usd(signal.entry_price.as_f64()), format!("\x1b[33mAPR {}\x1b[0m", signal.confidence), ] } @@ -78,13 +62,13 @@ 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") }, - format!("\x1b[35m{}\x1b[0m", signal.symbol), - signal.price.to_string(), + format_symbol(&signal.symbol), + format_usd(signal.entry_price.as_f64()), format!("\x1b[33mAPR {}\x1b[0m", signal.confidence), ] } @@ -95,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() { @@ -114,17 +98,17 @@ impl Formatted for MarketItem { impl Formatted for Position { fn get_formatted(&self) -> Vec { vec![ - self.symbol.to_string(), - format_f64(self.size), + format_symbol(&self.symbol), + 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![ @@ -247,7 +206,7 @@ impl Formatted for Vec { } } -impl Formatted for Strategy { +impl Formatted for StrategyStatus { fn get_formatted(&self) -> Vec { vec![ Triple( @@ -272,3 +231,70 @@ 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(); + + 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 +} 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<()> {