diff --git a/Cargo.lock b/Cargo.lock index eac0541..2242481 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3947,6 +3947,7 @@ version = "0.1.0-alpha.0" dependencies = [ "hypersdk", "postcard", + "rust_decimal", "serde", "tokio", ] @@ -3963,6 +3964,7 @@ dependencies = [ "pulse-sdk", "pulse-ui", "rand 0.8.7", + "rust_decimal", "serde", "serde_json", "tokio", diff --git a/Cargo.toml b/Cargo.toml index 9badfe3..c1d3ab1 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -11,6 +11,7 @@ hypersdk = { workspace = true } serde = { workspace = true } anyhow = { workspace = true } postcard = { workspace = true } +rust_decimal = { workspace = true } tokio = { workspace = true, features = [ "rt-multi-thread", "macros", @@ -29,6 +30,7 @@ toml = "1.1.3" members = ["pulse-ui", "pulse-sdk"] [workspace.dependencies] +rust_decimal = { version = "1.39", features = ["serde-str"] } postcard = { version = "1.1.3", features = ["alloc"] } pulse-ui = { path = "pulse-ui", version = "0.1.0-alpha.0" } pulse-sdk = { path = "pulse-sdk", version = "0.1.0-alpha.0" } diff --git a/pulse-sdk/Cargo.toml b/pulse-sdk/Cargo.toml index bb06f35..c545d4d 100644 --- a/pulse-sdk/Cargo.toml +++ b/pulse-sdk/Cargo.toml @@ -7,4 +7,5 @@ edition = "2024" serde = { workspace = true } hypersdk = { workspace = true } postcard = { workspace = true } -tokio = { workspace = true } +tokio = { workspace = true, features = ["io-std"] } +rust_decimal = { workspace = true } diff --git a/pulse-sdk/src/general.rs b/pulse-sdk/src/general.rs index a2fffad..f36fc66 100644 --- a/pulse-sdk/src/general.rs +++ b/pulse-sdk/src/general.rs @@ -1,3 +1,5 @@ +use rust_decimal::Decimal; + use crate::units::{Direction, Symbol, USD}; #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] @@ -20,7 +22,7 @@ pub struct Signal { pub symbol: String, pub kind: Direction, pub confidence: f32, - pub size: f64, + pub size: Decimal, pub price: USD, pub take_profit: USD, pub stop_loss: USD, diff --git a/pulse-sdk/src/lib.rs b/pulse-sdk/src/lib.rs index dd2211e..3ddcbb7 100644 --- a/pulse-sdk/src/lib.rs +++ b/pulse-sdk/src/lib.rs @@ -1,25 +1,114 @@ -#[cfg(target_os = "macos")] -use std::path::PathBuf; - pub mod general; -pub mod plugin; +pub mod strategy; pub mod terminal; pub mod units; pub use hypersdk; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; + +use crate::strategy::StrategyEngineMessage; + pub mod prelude { pub use crate::general::*; - pub use crate::plugin::*; + pub use crate::strategy::*; pub use crate::server_path; pub use crate::terminal::*; pub use crate::units::*; + pub use hypersdk; + pub use postcard; } -pub fn server_path() -> PathBuf { - PathBuf::from("/tmp/pulse-engine.sock") +pub fn server_path() -> std::path::PathBuf { + std::path::PathBuf::from("/tmp/pulse-engine.sock") } pub fn map_postcard_err(res: postcard::Result) -> tokio::io::Result { res.map_err(|e| tokio::io::Error::new(std::io::ErrorKind::Other, e)) } + +pub async fn send_raw(data: &[u8]) -> tokio::io::Result<()> { + let mut stdout = tokio::io::stdout(); + + stdout.write_all(&data.len().to_le_bytes()).await?; + stdout.write_all(data).await?; + stdout.flush().await?; + + 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 on_raw(&self, data: &[u8]) -> tokio::io::Result<()> { + match map_postcard_err(postcard::from_bytes(&data))? { + StrategyEngineMessage::Initialize => self.initialize().await, + StrategyEngineMessage::Command { command, args } => self.command(command, args).await, + StrategyEngineMessage::WatchList(watchlist) => self.watchlist(watchlist).await, + StrategyEngineMessage::Incoming(incoming) => self.incoming(incoming).await, + StrategyEngineMessage::Candlestick { + symbol, + interval, + candles, + } => self.candlestick(symbol, interval, candles).await, + } + } + + async fn initialize(&self) -> tokio::io::Result<()> { + unimplemented!("Strategy::initialize") + } + + async fn command(&self, _command: String, _args: Vec) -> tokio::io::Result<()> { + unimplemented!("Strategy::command") + } + + async fn watchlist(&self, _watchlist: Vec) -> tokio::io::Result<()> { + unimplemented!("Strategy::watchlist") + } + + async fn incoming(&self, _incoming: hypersdk::hypercore::Incoming) -> tokio::io::Result<()> { + unimplemented!("Strategy::event") + } + + async fn candlestick( + &self, + _symbol: String, + _interval: hypersdk::hypercore::CandleInterval, + _candles: Vec, + ) -> tokio::io::Result<()> { + unimplemented!("Strategy::candlestick") + } +} diff --git a/pulse-sdk/src/plugin.rs b/pulse-sdk/src/plugin.rs deleted file mode 100644 index 821a226..0000000 --- a/pulse-sdk/src/plugin.rs +++ /dev/null @@ -1,100 +0,0 @@ -use crate::{ - general::{EventLog, Signal}, - terminal::MarketItem, - units::Direction, -}; -use hypersdk::hypercore::{Candle, CandleInterval, Subscription}; - -#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] -pub struct StrategyManifest { - pub name: String, - pub description: String, - pub author: String, - pub version: String, -} - -#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] -pub struct RiskManifest { - pub name: String, - pub description: String, - pub author: String, - pub version: String, - - pub max_loss: u8, - pub cooldown: CandleInterval, -} - -#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] -pub enum StrategyMessage { - Log(EventLog), - - GetWatchList, - - GetCandlestick { - symbol: String, - interval: CandleInterval, - count: u32, - }, - - Subscribe(Subscription), - - Unsubscribe(Subscription), - - UnsubscribeAll, - - Signal(StrategySignal), -} - -#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] -pub enum RiskMessage { - Log(EventLog), - - GetWatchList, - - Approve(Signal), - - Reject { reason: String }, -} - -#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] -pub enum StrategyEngineMessage { - Initialize, - - WatchList(Vec), - - Command { - command: String, - args: Vec, - }, - - CandleUpdate { - symbol: String, - interval: CandleInterval, - candle: Candle, - }, - - Candlestick { - symbol: String, - interval: CandleInterval, - candles: Vec, - }, -} - -#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] -pub enum RiskEngineMessage { - Initialize, - - WatchList(Vec), - - Command { command: String, args: Vec }, - - Signal(StrategySignal), -} - -#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] -pub struct StrategySignal { - pub symbol: String, - pub side: Direction, - pub confidence: f32, - pub price: Option, -} diff --git a/pulse-sdk/src/strategy.rs b/pulse-sdk/src/strategy.rs new file mode 100644 index 0000000..291a480 --- /dev/null +++ b/pulse-sdk/src/strategy.rs @@ -0,0 +1,54 @@ +use crate::{ + general::{EventLog, Signal}, + terminal::MarketItem, +}; +use hypersdk::hypercore::{Candle, CandleInterval, Incoming, Subscription}; + +#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] +pub struct StrategyManifest { + pub name: String, + pub description: String, + pub author: String, + pub version: String, +} + +#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] +pub enum StrategyMessage { + Log(EventLog), + + GetWatchList, + + GetCandlestick { + symbol: String, + interval: CandleInterval, + count: u32, + }, + + Subscribe(Subscription), + + Unsubscribe(Subscription), + + UnsubscribeAll, + + Signal(Signal), +} + +#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] +pub enum StrategyEngineMessage { + Initialize, + + WatchList(Vec), + + Command { + command: String, + args: Vec, + }, + + Incoming(Incoming), + + Candlestick { + symbol: String, + interval: CandleInterval, + candles: Vec, + }, +} diff --git a/pulse-sdk/src/terminal.rs b/pulse-sdk/src/terminal.rs index a07e9f1..efac397 100644 --- a/pulse-sdk/src/terminal.rs +++ b/pulse-sdk/src/terminal.rs @@ -1,9 +1,11 @@ use crate::{ general::{EventLog, MarketTrend, Position, Signal}, - plugin::{RiskManifest, StrategyManifest}, + strategy::StrategyManifest, units::{Symbol, USD, Volatility}, }; -use hypersdk::hypercore::CandleInterval; +use hypersdk::{Decimal, hypercore::CandleInterval}; + +pub type SignalStatus = Result; #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] pub enum TerminalServerMessage { @@ -17,7 +19,7 @@ pub enum TerminalServerMessage { StrategyUpdated(Strategy), // Signals - SignalsUpdated(Vec), + SignalsUpdated(Vec), // Inspector Inspect(InspectTarget), @@ -39,7 +41,7 @@ pub enum TerminalClientMessage { pub struct MarketItem { pub symbol: Symbol, pub price: USD, - pub trend: f64, + pub trend: Decimal, pub volume_24h: USD, } @@ -108,7 +110,6 @@ pub enum ItemState { #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] pub struct Strategy { pub strategy: StrategyManifest, - pub risk: RiskManifest, pub mode: Mode, pub state: ItemState, diff --git a/pulse-sdk/src/units.rs b/pulse-sdk/src/units.rs index e2758e9..d598482 100644 --- a/pulse-sdk/src/units.rs +++ b/pulse-sdk/src/units.rs @@ -1,8 +1,10 @@ +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 f64); +pub struct USD(pub Decimal); #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] pub enum Direction { @@ -25,10 +27,10 @@ impl std::fmt::Display for Symbol { impl std::fmt::Display for USD { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - if self.0 > 0.0 { - write!(f, "\x1b[32m${}\x1b[0m", format_f64(self.0)) + 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)) + write!(f, "\x1b[31m${}\x1b[0m", format_f64(self.0.as_f64())) } } } diff --git a/src/engine/engine/command.rs b/src/engine/engine/command.rs index bd39825..1facea8 100644 --- a/src/engine/engine/command.rs +++ b/src/engine/engine/command.rs @@ -47,7 +47,7 @@ impl Engine { set_cfg!(id, { let id: String = id; - if !crate::store::pulse_plugin(&id)? + if !crate::store::pulse_strategy(&id)? .join("strategy.toml") .exists() { @@ -69,32 +69,6 @@ impl Engine { .await?; } - "risk" | "rs" => { - set_cfg!(id, { - let id: String = id; - - if !crate::store::pulse_plugin(&id)? - .join("risk.toml") - .exists() - { - return self - .terminal_server - .error( - "config::set::risk", - &format!("Non existent risk `{id}`"), - ) - .await; - } - - self.strategy.reload_risk(id.as_str()).await?; - self.config.lock().await.risk = id; - }); - - self.terminal_server - .info("config::set", "risk set successfully, use `config save` to persist changes") - .await?; - } - "cooldown" | "cool" | "cd" => { set_cfg!(cooldown, { self.config.lock().await.cooldown = cooldown; @@ -102,7 +76,7 @@ impl Engine { self.terminal_server.broadcast( pulse_sdk::terminal::TerminalServerMessage::StrategyUpdated(Strategy { strategy: self.strategy.strategy.manifest.lock().await.clone(), - risk: self.strategy.risk.manifest.lock().await.clone(), + mode: Mode::Auto, state: ItemState::Running, cooldown, @@ -114,7 +88,7 @@ impl Engine { _ => { self.terminal_server - .error("config::set", "Invalid usage, available options: watchlist, strategy, risk, cooldown") + .error("config::set", "Invalid usage, available options: watchlist, strategy, cooldown") .await?; } } diff --git a/src/engine/engine/execution.rs b/src/engine/engine/execution.rs new file mode 100644 index 0000000..9d56b4e --- /dev/null +++ b/src/engine/engine/execution.rs @@ -0,0 +1,98 @@ +use hypersdk::hypercore::{self, BatchOrder, OrderRequest, OrderTypePlacement, TimeInForce}; +use pulse_sdk::prelude::*; + +use crate::engine::Engine; + +impl Engine { + pub async fn execute_signal(&self, signal: &Signal) -> tokio::io::Result<()> { + let client = hypercore::mainnet(); + let accounts = self.accounts.lock().await; + + if let Some(acc) = accounts.get_active() { + let Some(asset_id) = self + .watch_list + .lock() + .await + .name_to_index + .get(&signal.symbol) + .cloned() + else { + self.terminal_server + .error( + "self::order", + &format!( + "Invalid Symbol: {:?}, Unable to get asset id", + signal.symbol + ), + ) + .await?; + + return Ok(()); + }; + + let order = BatchOrder { + orders: vec![ + OrderRequest { + asset: asset_id, + is_buy: matches!(signal.kind, Direction::Buy), + limit_px: signal.price.0, + sz: signal.size, + reduce_only: false, + order_type: OrderTypePlacement::Limit { + tif: TimeInForce::Gtc, + }, + cloid: Default::default(), + }, + OrderRequest { + asset: asset_id, + is_buy: matches!(signal.kind, Direction::Buy), + limit_px: signal.price.0, + sz: signal.size, + reduce_only: true, + order_type: OrderTypePlacement::Trigger { + is_market: true, + trigger_px: signal.take_profit.0, + 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, + reduce_only: true, + order_type: OrderTypePlacement::Trigger { + is_market: true, + trigger_px: signal.stop_loss.0, + tpsl: hypercore::TpSl::Sl, + }, + cloid: Default::default(), + }, + ], + grouping: hypercore::OrderGrouping::Na, + builder: None, + }; + + let nonce = chrono::Utc::now().timestamp_millis() as u64; + + match client + .place(&acc.private_key.0, order, nonce, None, None) + .await + { + Ok(_) => {} + Err(e) => { + self.terminal_server + .error("self::order", &e.to_string()) + .await?; + } + } + } else { + self.terminal_server + .error("Engine::order", "Unable to get active account") + .await?; + } + + Ok(()) + } +} diff --git a/src/engine/engine/mod.rs b/src/engine/engine/mod.rs index bb1d46f..f6cb30b 100644 --- a/src/engine/engine/mod.rs +++ b/src/engine/engine/mod.rs @@ -1,29 +1,37 @@ pub mod command; -pub mod plugin; +pub mod execution; +pub mod strategy; use crate::{ - engine::plugin::StrategyEngine, + engine::strategy::StrategyEngine, store::{accounts::AccountList, config::Config}, terminal::TerminalServer, }; use pulse_sdk::prelude::*; -use std::sync::Arc; +use std::{collections::HashMap, sync::Arc}; use tokio::{sync::Mutex, task::JoinHandle}; +#[derive(Debug, Clone)] +pub struct WatchList { + pub name_to_index: HashMap, + pub items: Vec, +} + #[derive(Clone)] pub struct Engine { pub terminal_server: Arc, pub strategy: Arc, pub config: Arc>, pub accounts: Arc>, - pub watch_list: Arc>>, + pub watch_list: Arc>, + pub signals: Arc>>, } impl Engine { pub async fn new() -> tokio::io::Result> { let config = Config::new().await?; - let strategy = StrategyEngine::new(&config.strategy, &config.risk).await?; + let strategy = StrategyEngine::new(&config.strategy).await?; let accounts = Arc::new(Mutex::new(AccountList::new().await?)); let config = Arc::new(Mutex::new(config)); @@ -33,7 +41,11 @@ impl Engine { strategy: strategy.initialize(engine.clone()), config, accounts, - watch_list: Arc::new(Mutex::new(Vec::new())), + watch_list: Arc::new(Mutex::new(WatchList { + name_to_index: HashMap::new(), + items: Vec::new(), + })), + signals: Arc::new(Mutex::new(Vec::new())), })) } @@ -59,11 +71,7 @@ impl Engine { if let Err(error) = self .terminal_server - .broadcast( - pulse_sdk::terminal::TerminalServerMessage::WatchListUpdated( - watch_list, - ), - ) + .broadcast(TerminalServerMessage::WatchListUpdated(watch_list.items)) .await { self.terminal_server @@ -91,24 +99,21 @@ impl Engine { match client.clearinghouse_state(acc.address, None).await { Ok(state) => { self.terminal_server - .broadcast( - pulse_sdk::terminal::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 - .map(|px| px.as_f64()) - .unwrap_or(0.0)), - profit: USD(position.position.unrealized_pnl.as_f64()), - }) - .collect(), - ), - ) + .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) => { diff --git a/src/engine/engine/plugin.rs b/src/engine/engine/strategy.rs similarity index 66% rename from src/engine/engine/plugin.rs rename to src/engine/engine/strategy.rs index 2c30d1b..8f8f074 100644 --- a/src/engine/engine/plugin.rs +++ b/src/engine/engine/strategy.rs @@ -15,12 +15,11 @@ use tokio::{ use crate::{ engine::Engine, - store::{plugin::Plugin, pulse_plugin}, + store::{pulse_strategy, strategy::StrategyChild}, }; pub struct StrategyEngine { - pub strategy: Arc>, - pub risk: Arc>, + pub strategy: Arc, pub engine: Weak, pub ws: WebSocket, @@ -28,25 +27,17 @@ pub struct StrategyEngine { } impl StrategyEngine { - pub async fn new(strategy_id: &str, risk_id: &str) -> tokio::io::Result { - let strategy = pulse_plugin(strategy_id)?; - let risk = pulse_plugin(risk_id)?; + pub async fn new(strategy_id: &str) -> tokio::io::Result { + let strategy = pulse_strategy(strategy_id)?; - let (strategy, strategy_manifest) = get_manifest_plugin_pair( + let (strategy, strategy_manifest) = get_manifest( &strategy, &strategy.join("strategy.bash"), &fs::read(strategy.join("strategy.toml")).await?, )?; - let (risk, risk_manifest) = get_manifest_plugin_pair( - &risk, - &risk.join("risk.bash"), - &fs::read(risk.join("risk.toml")).await?, - )?; - Ok(Self { - strategy: Arc::new(Plugin::new(strategy, strategy_manifest)), - risk: Arc::new(Plugin::new(risk, risk_manifest)), + strategy: Arc::new(StrategyChild::new(strategy, strategy_manifest)), engine: Weak::new(), ws: hypercore::mainnet_ws(), subscriptions: Mutex::new(HashSet::new()), @@ -76,7 +67,7 @@ impl StrategyEngine { Some(StrategyMessage::GetWatchList) => { self.strategy .send(&StrategyEngineMessage::WatchList( - engine.watch_list.lock().await.clone(), + engine.watch_list.lock().await.clone().items, )) .await?; } @@ -87,7 +78,24 @@ impl StrategyEngine { } Some(StrategyMessage::Signal(signal)) => { - self.risk.send(&RiskEngineMessage::Signal(signal)).await?; + match engine.execute_signal(&signal).await { + Ok(_) => engine.signals.lock().await.push(Ok(signal)), + Err(e) => { + engine.signals.lock().await.push(Err(signal)); + + engine + .terminal_server + .error("signal", &format!("Failed to execute signal: {e}")) + .await? + } + } + + engine + .terminal_server + .broadcast(TerminalServerMessage::SignalsUpdated( + engine.signals.lock().await.clone(), + )) + .await?; } Some(StrategyMessage::Subscribe(subscription)) => { @@ -153,54 +161,19 @@ impl StrategyEngine { } } - pub async fn run_risk(&self) -> tokio::io::Result<()> { - let engine = self - .engine - .upgrade() - .expect("Failed to upgrade engine (StrategyEngine)"); - - self.risk.send(&RiskEngineMessage::Initialize).await?; - - loop { - match self.risk.recv().await? { - None => {} - - Some(RiskMessage::GetWatchList) => { - self.risk - .send(&&RiskEngineMessage::WatchList( - engine.watch_list.lock().await.clone(), - )) - .await?; - } - - Some(RiskMessage::Log(mut log)) => { - log.name.insert_str(0, "risk::"); - engine.terminal_server.log_raw(log).await?; - } - - Some(RiskMessage::Approve(signal)) => {} - Some(RiskMessage::Reject { reason }) => {} - } - } - } - pub async fn spawn(self: &Arc) { let engine = self.clone(); tokio::spawn(async move { engine.run_strategy().await }); - - let engine = self.clone(); - - tokio::spawn(async move { engine.run_risk().await }); } pub async fn reload_strategy(self: &Arc, id: &str) -> tokio::io::Result<()> { - let plugin = pulse_plugin(id)?; + let strategy = pulse_strategy(id)?; - let (child, manifest) = get_manifest_plugin_pair( - &plugin, - &plugin.join("strategy.bash"), - &fs::read(plugin.join("strategy.toml")).await?, + let (child, manifest) = get_manifest( + &strategy, + &strategy.join("strategy.bash"), + &fs::read(strategy.join("strategy.toml")).await?, )?; self.strategy.reload(child, manifest).await?; @@ -210,34 +183,17 @@ impl StrategyEngine { Ok(()) } - - pub async fn reload_risk(self: &Arc, id: &str) -> tokio::io::Result<()> { - let plugin = pulse_plugin(id)?; - - let (child, manifest) = get_manifest_plugin_pair( - &plugin, - &plugin.join("risk.bash"), - &fs::read(plugin.join("risk.toml")).await?, - )?; - - self.risk.reload(child, manifest).await?; - - let engine = self.clone(); - tokio::spawn(async move { engine.run_risk().await }); - - Ok(()) - } } -pub fn get_manifest_plugin_pair<'de, M: serde::Deserialize<'de>>( - plugin_dir: &PathBuf, - plugin_path: &PathBuf, +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(plugin_path) - .current_dir(plugin_dir) + .arg(strategy_path) + .current_dir(strategy_dir) .stdin(Stdio::piped()) .stdout(Stdio::piped()) .stderr(Stdio::inherit()) diff --git a/src/engine/fetch.rs b/src/engine/fetch.rs index d8be404..bbd6038 100644 --- a/src/engine/fetch.rs +++ b/src/engine/fetch.rs @@ -1,20 +1,23 @@ +use rust_decimal::Decimal; use pulse_sdk::prelude::*; use serde_json::Value; use std::collections::HashMap; -fn number(value: &Value, field: &str) -> Result { +use crate::engine::WatchList; + +fn number(value: &Value, field: &str) -> Result { let raw = value[field] .as_str() .ok_or_else(|| format!("asset context field {field} must be a string"))?; - raw.parse::() + raw.parse::() .map_err(|error| format!("could not parse asset context field {field} ({raw}): {error}")) } pub async fn fetch_watch_list( client: &hypersdk::hypercore::HttpClient, symbols: &[String], -) -> Result, String> { +) -> Result { let response = client .meta_and_asset_ctxs(None) .await @@ -34,6 +37,7 @@ pub async fn fetch_watch_list( let universe = response[0]["universe"] .as_array() .ok_or("metaAndAssetCtxs response is missing meta.universe")?; + let contexts = response[1] .as_array() .ok_or("metaAndAssetCtxs response contexts must be an array")?; @@ -46,37 +50,46 @@ pub async fn fetch_watch_list( )); } - let mut by_symbol = HashMap::with_capacity(universe.len()); + // Build symbol -> perp asset index + let mut name_to_index = HashMap::with_capacity(universe.len()); - for (meta, context) in universe.iter().zip(contexts) { + for (index, meta) in universe.iter().enumerate() { let symbol = meta["name"] .as_str() .ok_or("instrument metadata is missing a name")?; + + name_to_index.insert(symbol.to_owned(), index); + } + + // Now only process requested symbols + let mut items = Vec::with_capacity(symbols.len()); + + for symbol in symbols { + let index = *name_to_index + .get(symbol) + .ok_or_else(|| format!("{symbol} is not in the Hyperliquid perpetual universe"))?; + + let context = &contexts[index]; + let price = number(context, "markPx")?; let previous_day_price = number(context, "prevDayPx")?; let volume_24h = number(context, "dayNtlVlm")?; - if previous_day_price <= 0.0 { + if previous_day_price.is_zero() || previous_day_price.is_sign_negative() { return Err(format!("{symbol} has an invalid previous-day price")); } - by_symbol.insert( - symbol, - MarketItem { - symbol: Symbol(symbol.to_owned()), - price: USD(price), - volume_24h: USD(volume_24h), - trend: ((price / previous_day_price) - 1.0) * 100.0, - }, - ); + items.push(MarketItem { + symbol: Symbol(symbol.clone()), + price: USD(price), + volume_24h: USD(volume_24h), + trend: ((price / previous_day_price) - >::from(1)) + * >::from(100), + }); } - symbols - .iter() - .map(|symbol| { - by_symbol - .remove(symbol.as_str()) - .ok_or_else(|| format!("{symbol} is not in the Hyperliquid perpetual universe")) - }) - .collect() + Ok(WatchList { + items, + name_to_index, + }) } diff --git a/src/engine/store/config.rs b/src/engine/store/config.rs index c16d916..ba42aa2 100644 --- a/src/engine/store/config.rs +++ b/src/engine/store/config.rs @@ -4,7 +4,6 @@ use hypersdk::hypercore::CandleInterval; pub struct Config { pub watchlist: Vec, pub strategy: String, - pub risk: String, pub cooldown: CandleInterval, } @@ -13,7 +12,6 @@ impl Default for Config { Self { watchlist: vec!["BTC".to_string(), "SOL".to_string(), "ETH".to_string()], strategy: String::new(), - risk: String::new(), cooldown: CandleInterval::ThirtyMinutes, } } diff --git a/src/engine/store/mod.rs b/src/engine/store/mod.rs index 73b6dee..7552bd3 100644 --- a/src/engine/store/mod.rs +++ b/src/engine/store/mod.rs @@ -2,7 +2,7 @@ use std::path::PathBuf; pub mod accounts; pub mod config; -pub mod plugin; +pub mod strategy; pub fn home_dir() -> tokio::io::Result { std::env::home_dir().ok_or_else(|| { @@ -14,12 +14,12 @@ pub fn pulse_directory() -> tokio::io::Result { Ok(home_dir()?.join(".config").join("pulse-trader")) } -pub fn pulse_plugins_directory() -> tokio::io::Result { - Ok(pulse_directory()?.join("plugins")) +pub fn pulse_strategies_directory() -> tokio::io::Result { + Ok(pulse_directory()?.join("strategies")) } -pub fn pulse_plugin(id: &str) -> tokio::io::Result { - Ok(pulse_directory()?.join("plugins").join(id)) +pub fn pulse_strategy(id: &str) -> tokio::io::Result { + Ok(pulse_directory()?.join("strategies").join(id)) } pub fn pulse_config_file() -> tokio::io::Result { diff --git a/src/engine/store/plugin.rs b/src/engine/store/strategy.rs similarity index 69% rename from src/engine/store/plugin.rs rename to src/engine/store/strategy.rs index 62449f6..fa7a99f 100644 --- a/src/engine/store/plugin.rs +++ b/src/engine/store/strategy.rs @@ -1,33 +1,29 @@ -use std::marker::PhantomData; - -use pulse_sdk::map_postcard_err; -use serde::{Deserialize, Serialize}; +use pulse_sdk::{ + map_postcard_err, + strategy::{StrategyEngineMessage, StrategyManifest, StrategyMessage}, +}; use tokio::{ io::{AsyncReadExt, AsyncWriteExt}, process::{Child, ChildStdout}, sync::Mutex, }; #[derive(Debug)] -pub struct Plugin Deserialize<'de>, M: for<'de> Deserialize<'de>> { - pub manifest: Mutex, +pub struct StrategyChild { + pub manifest: Mutex, pub stdout: Mutex, pub process: Mutex, - - pub _p: (PhantomData, PhantomData), } -impl Deserialize<'de>, M: for<'de> Deserialize<'de>> Plugin { - pub fn new(mut child: Child, manifest: M) -> Self { +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), - - _p: (PhantomData, PhantomData), } } - pub async fn recv(&self) -> tokio::io::Result> { + pub async fn recv(&self) -> tokio::io::Result> { let mut stdout = self.stdout.lock().await; let mut len_buf = [0u8; size_of::()]; @@ -46,7 +42,7 @@ impl Deserialize<'de>, M: for<'de> Deserialize<'de>> P Ok(Some(map_postcard_err(postcard::from_bytes(&buffer))?)) } - pub async fn send(&self, msg: &S) -> tokio::io::Result<()> { + pub async fn send(&self, msg: &StrategyEngineMessage) -> tokio::io::Result<()> { self.send_raw(&map_postcard_err(postcard::to_allocvec(msg))?) .await } @@ -62,7 +58,11 @@ impl Deserialize<'de>, M: for<'de> Deserialize<'de>> P Ok(()) } - pub async fn reload(&self, mut child: Child, manifest: M) -> tokio::io::Result<()> { + pub async fn reload( + &self, + mut child: Child, + manifest: StrategyManifest, + ) -> tokio::io::Result<()> { let mut process = self.process.lock().await; process.kill().await?; diff --git a/src/engine/terminal.rs b/src/engine/terminal.rs index ba14436..97e8150 100644 --- a/src/engine/terminal.rs +++ b/src/engine/terminal.rs @@ -66,7 +66,6 @@ impl TerminalServer { id, pulse_sdk::terminal::TerminalServerMessage::StrategyUpdated(Strategy { strategy: engine.strategy.strategy.manifest.lock().await.clone(), - risk: engine.strategy.risk.manifest.lock().await.clone(), mode: Mode::Auto, state: ItemState::Running, cooldown: engine.config.lock().await.cooldown, @@ -164,8 +163,8 @@ impl TerminalServer { } pub async fn send_to_client(client: &mut OwnedWriteHalf, msg: &[u8]) -> tokio::io::Result<()> { - client.write(&msg.len().to_le_bytes()).await?; - client.write(msg).await?; + client.write_all(&msg.len().to_le_bytes()).await?; + client.write_all(msg).await?; client.flush().await?; Ok(()) diff --git a/src/terminal/formatting.rs b/src/terminal/formatting.rs index 50211d6..dace9dd 100644 --- a/src/terminal/formatting.rs +++ b/src/terminal/formatting.rs @@ -58,17 +58,37 @@ impl Formatted for EventLog { } } -impl Formatted for Signal { +impl Formatted for SignalStatus { fn get_formatted(&self) -> Vec { - vec![ - if matches!(self.kind, Direction::Buy) { - format!("\x1b[32m{}\x1b[0m", self.kind) - } else { - format!("\x1b[31m{}\x1b[0m", self.kind) - }, - format!("\x1b[35m{}\x1b[0m", self.symbol), - self.price.to_string(), - ] + match self { + Ok(signal) => { + vec![ + format!("\x1b[33mOK\x1b[0m"), + if matches!(signal.kind, Direction::Buy) { + format!("\x1b[32mBUY\x1b[0m") + } else { + format!("\x1b[31mSELL\x1b[0m") + }, + format!("\x1b[35m{}\x1b[0m", signal.symbol), + signal.price.to_string(), + format!("\x1b[33mAPR {}\x1b[0m", signal.confidence), + ] + } + + Err(signal) => { + vec![ + format!("\x1b[31mERR\x1b[0m"), + if matches!(signal.kind, Direction::Buy) { + format!("\x1b[32mBUY\x1b[0m") + } else { + format!("\x1b[31mSELL\x1b[0m") + }, + format!("\x1b[35m{}\x1b[0m", signal.symbol), + signal.price.to_string(), + format!("\x1b[33mAPR {}\x1b[0m", signal.confidence), + ] + } + } } } @@ -85,7 +105,7 @@ impl Formatted for MarketItem { } else { "\x1b[31m▼\x1b[0m" }, - format_f64(self.trend.abs()) + format_f64(self.trend.as_f64().abs()) ), ] } @@ -232,38 +252,13 @@ impl Formatted for Strategy { vec![ Triple( "\x1b[2mStrategy\x1b[0m", - "\x1b[2mRisk\x1b[0m", - "\x1b[2mStrat Ver\x1b[0m", + "\x1b[2mState\x1b[0m", + "\x1b[2mMode\x1b[0m", ), Triple( &format!("\x1b[97m{}\x1b[0m", self.strategy.name), - &format!("\x1b[93m{}\x1b[0m", self.risk.name), - &format!("\x1b[90m{}\x1b[0m", self.strategy.version), - ), - Triple("", "", ""), - Triple( - "\x1b[2mMode\x1b[0m", - "\x1b[2mState\x1b[0m", - "\x1b[2mRisk Ver\x1b[0m", - ), - Triple( - &self.mode.to_string(), &self.state.to_string(), - &format!("\x1b[90m{}\x1b[0m", self.risk.version), - ), - Triple("", "", ""), - Triple( - "\x1b[2mCooldown\x1b[0m", - "\x1b[2mStrat Author\x1b[0m", - "\x1b[2mMax loss\x1b[0m", - ), - Triple( - &format!( - "\x1b[96m{}\x1b[0m (\x1b[90m{} rec\x1b[0m)", - self.cooldown, self.risk.cooldown - ), - &format!("\x1b[96m{:?}\x1b[0m", self.strategy.author), - &format!("\x1b[93m{}%\x1b[0m", self.risk.max_loss), + &self.mode.to_string(), ), ] .get_formatted() diff --git a/src/terminal/main.rs b/src/terminal/main.rs index 25ae2f5..e751356 100644 --- a/src/terminal/main.rs +++ b/src/terminal/main.rs @@ -35,7 +35,7 @@ pub struct PulseTradeApp { watch_list: State>, active_positions: State>, logs: State>, - signals: State>, + signals: State>, inspect: State, } diff --git a/src/terminal/terminal.rs b/src/terminal/terminal.rs index 3657ce2..608d1b8 100644 --- a/src/terminal/terminal.rs +++ b/src/terminal/terminal.rs @@ -31,8 +31,8 @@ impl TerminalClient { message: pulse_sdk::terminal::TerminalClientMessage, ) -> tokio::io::Result<()> { let msg = map_postcard_err(postcard::to_allocvec(&message))?; - self.writer.write(&msg.len().to_le_bytes()).await?; - self.writer.write(&msg).await?; + self.writer.write_all(&msg.len().to_le_bytes()).await?; + self.writer.write_all(&msg).await?; self.writer.flush().await?; Ok(()) @@ -53,16 +53,22 @@ impl TerminalClient { let status = app.status.clone(); let inspect = app.inspect.clone(); - tokio::spawn(Self::run_client( - reader, - watch_list, - active_positions, - logs, - signals, - market_overview, - status, - inspect, - )); + tokio::spawn(async { + if let Err(v) = Self::run_client( + reader, + watch_list, + active_positions, + logs, + signals, + market_overview, + status, + inspect, + ) + .await + { + panic!("{v}") + } + }); app.sock = Some(self); @@ -75,60 +81,60 @@ impl TerminalClient { watch_list: State>, active_positions: State>, logs: State>, - signals: State>, + signals: State>, market_overview: State>, status: State>, inspect: State, ) -> tokio::io::Result<()> { - let mut len_buf = [0u8; size_of::()]; - reader - .read_exact(&mut len_buf) - .await - .expect("Failed to get header length"); + loop { + let mut len_buf = [0u8; size_of::()]; + reader + .read_exact(&mut len_buf) + .await + .expect("Failed to get header length"); - let len = usize::from_le_bytes(len_buf); + let len = usize::from_le_bytes(len_buf); - let mut buffer = vec![0u8; len]; + let mut buffer = vec![0u8; len]; - reader - .read_exact(&mut buffer) - .await - .expect("Failed to read socket"); + reader + .read_exact(&mut buffer) + .await + .expect("Failed to read socket"); - match map_postcard_err(postcard::from_bytes(&buffer))? { - TerminalServerMessage::WatchListUpdated(v) => { - *watch_list.lock().await = v; - } + match map_postcard_err(postcard::from_bytes(&buffer))? { + TerminalServerMessage::WatchListUpdated(v) => { + *watch_list.lock().await = v; + } - TerminalServerMessage::PositionsUpdated(v) => { - *active_positions.lock().await = v; - } + TerminalServerMessage::PositionsUpdated(v) => { + *active_positions.lock().await = v; + } - TerminalServerMessage::StrategyUpdated(v) => { - *market_overview.lock().await = Some(v); - } + TerminalServerMessage::StrategyUpdated(v) => { + *market_overview.lock().await = Some(v); + } - TerminalServerMessage::SignalsUpdated(v) => { - *signals.lock().await = v; - } + TerminalServerMessage::SignalsUpdated(v) => { + *signals.lock().await = v; + } - TerminalServerMessage::Inspect(v) => { - *inspect.lock().await = v; - } + TerminalServerMessage::Inspect(v) => { + *inspect.lock().await = v; + } - TerminalServerMessage::StatusUpdated(v) => { - *status.lock().await = Some(v); - } + TerminalServerMessage::StatusUpdated(v) => { + *status.lock().await = Some(v); + } - TerminalServerMessage::SetLogs(v) => { - *logs.lock().await = v; - } + TerminalServerMessage::SetLogs(v) => { + *logs.lock().await = v; + } - TerminalServerMessage::AddLog(v) => { - logs.lock().await.push(v); + TerminalServerMessage::AddLog(v) => { + logs.lock().await.push(v); + } } } - - Ok(()) } }