diff --git a/pulse-sdk/src/general.rs b/pulse-sdk/src/general.rs index 03582af..ec178ec 100644 --- a/pulse-sdk/src/general.rs +++ b/pulse-sdk/src/general.rs @@ -1,4 +1,7 @@ -use hypersdk::hypercore::Side; +use hypersdk::{ + dec, + hypercore::{Cloid, Side}, +}; use rust_decimal::Decimal; #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] @@ -30,14 +33,31 @@ pub enum Mode { Manual, } +#[derive(Debug, Clone, Copy)] +pub enum Allocation { + Fixed(Decimal), + Percent(Decimal), +} + #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] pub struct Signal { pub symbol: String, pub side: Side, - pub confidence: f32, + /// 0-100 + pub confidence: u8, pub entry_price: Decimal, pub take_profit: Decimal, +} + +#[derive(Debug, Clone, Copy, serde::Serialize, serde::Deserialize)] +pub struct EngineOrder { + pub risk: Decimal, + pub size: Decimal, pub stop_loss: Decimal, + + pub entry: Cloid, + pub take_profit: Cloid, + pub stop_loss_id: Cloid, } #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] @@ -57,8 +77,10 @@ pub struct Position { #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] pub struct EngineStatus { + pub perps_equity: Decimal, pub strategy_mode: Mode, pub strategy_state: ItemState, + pub positions: Vec, } impl std::fmt::Display for MarketTrend { @@ -101,3 +123,75 @@ impl std::fmt::Display for ItemState { } } } + +impl<'de> serde::Deserialize<'de> for Allocation { + fn deserialize(deserializer: D) -> Result + where + D: serde::Deserializer<'de>, + { + struct AllocationVisitor; + + impl<'de> serde::de::Visitor<'de> for AllocationVisitor { + type Value = Allocation; + + fn expecting(&self, formatter: &mut std::fmt::Formatter) -> std::fmt::Result { + formatter.write_str("a number or percentage string") + } + + fn visit_f64(self, value: f64) -> Result + where + E: serde::de::Error, + { + Ok(Allocation::Fixed(Decimal::from_f64_retain(value).unwrap())) + } + + fn visit_str(self, value: &str) -> Result + where + E: serde::de::Error, + { + if let Some(percent) = value.strip_suffix('%') { + let value = percent.parse::().map_err(E::custom)?; + + Ok(Allocation::Percent(value)) + } else { + Err(E::custom("invalid allocation format")) + } + } + } + + deserializer.deserialize_any(AllocationVisitor) + } +} + +impl serde::Serialize for Allocation { + fn serialize(&self, serializer: S) -> Result + where + S: serde::Serializer, + { + match self { + Allocation::Fixed(value) => serde::Serialize::serialize(value, serializer), + Allocation::Percent(value) => { + let s = format!("{}%", value); + serializer.serialize_str(&s) + } + } + } +} + +impl Allocation { + pub fn get(&self, full_alloc: Decimal) -> Decimal { + match self { + Self::Fixed(f) => *f, + Self::Percent(p) => (*p / dec!(100.0)) * full_alloc, + } + } +} + +impl std::fmt::Display for Allocation { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Self::Fixed(v) => write!(f, "${v}"), + Self::Percent(p) => write!(f, "{p}%"), + } + } +} diff --git a/pulse-sdk/src/lib.rs b/pulse-sdk/src/lib.rs index af0744e..340dfb9 100644 --- a/pulse-sdk/src/lib.rs +++ b/pulse-sdk/src/lib.rs @@ -3,6 +3,8 @@ pub mod strategy; pub mod terminal; pub use hypersdk; +use hypersdk::Decimal; +use rust_decimal::RoundingStrategy; use tokio::io::{AsyncReadExt, AsyncWriteExt}; use crate::{general::LogKind, strategy::StrategyEngineMessage}; @@ -10,9 +12,9 @@ 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::{round_price, server_path}; pub use hypersdk; pub use postcard; @@ -26,6 +28,21 @@ 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 fn round_price(price: Decimal) -> Decimal { + let digits = price.trunc().to_string().len() as u32; + let decimal_places = 5_i32 - digits as i32; + + if decimal_places < 0 { + let factor = Decimal::from(10_u64.pow((-decimal_places) as u32)); + (price / factor).round() * factor + } else { + price.round_dp_with_strategy( + decimal_places as u32, + RoundingStrategy::MidpointAwayFromZero, + ) + } +} + pub async fn send_raw(data: &[u8]) -> tokio::io::Result<()> { let mut stdout = tokio::io::stdout(); diff --git a/pulse-sdk/src/terminal.rs b/pulse-sdk/src/terminal.rs index c82562b..0063acb 100644 --- a/pulse-sdk/src/terminal.rs +++ b/pulse-sdk/src/terminal.rs @@ -1,19 +1,16 @@ use crate::{ - general::{EngineStatus, EventLog, Position, Signal}, + general::{EngineOrder, EngineStatus, EventLog, Signal}, strategy::StrategyManifest, }; use hypersdk::Decimal; -pub type SignalStatus = Result; +pub type SignalStatus = Result<(Signal, Option), (Signal, Option)>; #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] pub enum TerminalServerMessage { // WatchList WatchListUpdated(Vec), - // Positions - PositionsUpdated(Vec), - // Configuration ConfigUpdated(EngineConfig), @@ -63,5 +60,10 @@ pub enum InspectItem { #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] pub struct EngineConfig { + pub preset: String, pub strategy: StrategyManifest, + pub description: Option, + + pub risk_per_trade: String, + pub max_daily_loss: String, } diff --git a/src/engine/engine/command.rs b/src/engine/engine/command.rs index 0a815e1..07bf9e2 100644 --- a/src/engine/engine/command.rs +++ b/src/engine/engine/command.rs @@ -1,7 +1,6 @@ -use crate::{engine::Engine, store::config::Config}; +use crate::{engine::Engine, store::config::ConfigManager}; use pulse_sdk::prelude::*; -use toml::Value; impl Engine { pub async fn invalid_command_usage(&self, name: &str) -> tokio::io::Result<()> { @@ -13,7 +12,44 @@ impl Engine { command: &str, mut args: Vec<&str>, ) -> tokio::io::Result<()> { + macro_rules! set_cfg { + ($n:ident) => { + let Ok(Ok($n)) = toml::value::Value::try_from(args[2]).map(|v| v.try_into()) else { + return self + .terminal_server + .error("config::set", "Unable to parse value") + .await; + }; + }; + } + match command { + "help" | "?" => { + self.terminal_server + .info("engine::help", "AVAILABLE COMMANDS") + .await?; + + let commands = [ + ("help", "Show this help message"), + ("config reload", "Reload configuration from disk"), + ("config save", "Save current configuration"), + ("account list", "List available accounts"), + ("account use ", "Switch active account"), + ("preset list", "List available presets"), + ("preset use ", "Switch active preset"), + ("strategy start", "Start strategy runtime"), + ("strategy restart", "Restart strategy runtime"), + ("strategy set ", "Change active strategy"), + ("strategy ", "Send command to strategy runtime"), + ]; + + for (command, description) in commands { + self.terminal_server + .info("engine::help", &format!("{:<30} {}", command, description)) + .await?; + } + } + "config" | "cfg" => { if args.len() == 0 { return self.invalid_command_usage("config").await; @@ -21,10 +57,23 @@ impl Engine { match args[0] { "reload" => { - *self.config.lock().await = Config::new().await?; + let mut config = self.config.lock().await; + + let new_config = ConfigManager::new().await?; + let strategy_changed = + new_config.get_ref()?.strategy != config.get_ref()?.strategy; + + *config = new_config; + self.terminal_server .info("engine::config", "successfully reloaded") .await?; + + if strategy_changed { + self.strategy_engine + .reload(&config.get_ref()?.strategy) + .await?; + } } "save" => { @@ -34,90 +83,6 @@ impl Engine { .await?; } - "set" => { - macro_rules! set_cfg { - ($n:ident, $v:expr) => { - if let Ok(Ok($n)) = Value::try_from(args[2]).map(|v| v.try_into()) { - $v; - } else { - self.terminal_server - .error("config::set", "Unable to parse value") - .await?; - } - }; - } - - if args.len() == 3 { - match args[1] { - "watchlist" | "watch" | "wl" => { - set_cfg!(v, self.config.lock().await.watchlist = v); - - self.terminal_server - .info("config::set", "watchlist set successfully, use `config save` to persist changes") - .await?; - - self.terminal_server - .broadcast(TerminalServerMessage::ConfigUpdated( - self.get_config_status().await, - )) - .await?; - } - - "strategy" | "strat" | "sg" => { - set_cfg!(id, { - let id: String = id; - - if !crate::store::pulse_strategy(&id)? - .join("strategy.toml") - .exists() - { - return self - .terminal_server - .error( - "config::set[strategy]", - &format!("Non existent strategy `{id}`"), - ) - .await; - } - - self.strategy_engine.reload(id.as_str()).await?; - self.config.lock().await.strategy = id; - }); - - self.terminal_server - .info("config::set", "strategy set successfully, use `config save` to persist changes") - .await?; - - self.terminal_server - .broadcast(TerminalServerMessage::ConfigUpdated( - self.get_config_status().await, - )) - .await?; - } - - "cooldown" | "cool" | "cd" => { - set_cfg!(cooldown, { - self.config.lock().await.cooldown = cooldown; - }); - - self.terminal_server - .broadcast(TerminalServerMessage::ConfigUpdated( - self.get_config_status().await, - )) - .await?; - } - - _ => { - self.terminal_server - .error("config::set", "Invalid usage, available options: watchlist, strategy, cooldown") - .await?; - } - } - } else { - self.invalid_command_usage("engine::config").await?; - } - } - _ => { self.invalid_command_usage("engine::config").await?; } @@ -169,7 +134,7 @@ impl Engine { .terminal_server .error( "engine::account", - &format!("Account not found ({new_active})"), + &format!("account not found ({new_active})"), ) .await; } @@ -179,7 +144,7 @@ impl Engine { self.terminal_server .info( "engine::account", - &format!("Account set to {new_active} successfully!"), + &format!("account set to {new_active} successfully!"), ) .await?; } @@ -190,6 +155,80 @@ impl Engine { } } + "preset" => { + if args.len() == 0 { + return self.invalid_command_usage("engine::preset").await; + } + + match args[0] { + "list" | "ls" => { + self.terminal_server + .info("engine::preset", "PRESET LIST") + .await?; + + let config = self.config.lock().await; + + for (name, cfg) in &config.configs { + let description = if let Some(desc) = &cfg.description { + format!("- {desc}") + } else { + String::new() + }; + + self.terminal_server + .info( + "engine::preset", + &if name == &config.preset { + format!("{name} (active) {description}") + } else { + format!("{name} {description}") + }, + ) + .await?; + } + } + + "use" | "set" => { + if args.len() < 2 { + return self.invalid_command_usage("engine::preset").await; + } + + let new_active = args[1]; + + let mut config = self.config.lock().await; + + if !config.configs.contains_key(new_active) { + return self + .terminal_server + .error( + "engine::preset", + &format!("preset not found ({new_active})"), + ) + .await; + } + + config.preset = new_active.to_string(); + + self.terminal_server + .info( + "engine::preset", + &format!("preset set to {new_active} successfully!"), + ) + .await?; + + self.terminal_server + .broadcast(TerminalServerMessage::ConfigUpdated( + self.get_config_status().await?, + )) + .await?; + } + + _ => { + return self.invalid_command_usage("engine::preset").await; + } + } + } + "strategy" | "strat" | "sg" => { let Some(strategy_command) = args.drain(0..=0).next() else { return self.invalid_command_usage("strategy").await; @@ -198,6 +237,41 @@ impl Engine { match strategy_command { "start" => {} + "restart" => { + self.strategy_engine + .reload(&self.config.lock().await.get_ref()?.strategy) + .await?; + } + + "set" => { + set_cfg!(id); + + let id: String = id; + + if !crate::store::pulse_strategy(&id)? + .join("strategy.toml") + .exists() + { + return self + .terminal_server + .error( + "config::set[strategy]", + &format!("Non existent strategy `{id}`"), + ) + .await; + } + + self.strategy_engine.reload(id.as_str()).await?; + self.config.lock().await.get_mut()?.strategy = id; + + self.terminal_server + .info( + "config::set", + "strategy set successfully, use `config save` to persist changes", + ) + .await?; + } + _ => { self.strategy_engine .send(&StrategyEngineMessage::Command { @@ -211,11 +285,20 @@ impl Engine { _ => { self.terminal_server - .error("engine::cmd", &format!("Command '{}' not found", command)) + .error( + "engine::cmd", + &format!("command '{}' not found, use help or ?", command), + ) .await?; } } + self.terminal_server + .broadcast(TerminalServerMessage::ConfigUpdated( + self.get_config_status().await?, + )) + .await?; + Ok(()) } } diff --git a/src/engine/engine/execution.rs b/src/engine/engine/execution.rs index a7370c0..96279ec 100644 --- a/src/engine/engine/execution.rs +++ b/src/engine/engine/execution.rs @@ -1,96 +1,122 @@ -use hypersdk::hypercore::{self, BatchOrder, OrderRequest, OrderTypePlacement, Side, TimeInForce}; +use hypersdk::hypercore::{ + self, BatchOrder, NonceHandler, OrderRequest, OrderResponseStatus, OrderTypePlacement, Side, + TimeInForce, +}; use pulse_sdk::prelude::*; use crate::engine::Engine; impl Engine { - pub async fn execute_signal(&self, signal: &Signal) -> tokio::io::Result<()> { + pub async fn execute_signal( + &self, + size: &mut Option, + signal: &Signal, + ) -> anyhow::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?; + let acc = accounts + .get_active() + .ok_or_else(|| anyhow::anyhow!("Unable to get active account"))?; - return Ok(()); - }; + let Some(asset_id) = self + .watch_list + .lock() + .await + .name_to_index + .get(&signal.symbol) + .cloned() + else { + return Err(anyhow::anyhow!( + "Invalid Symbol: {:?}, Unable to get asset id", + signal.symbol + )); + }; - let order = BatchOrder { - orders: vec![ - OrderRequest { - asset: asset_id, - 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, - }, - cloid: Default::default(), + let engine_order = self.risk_engine.create_order(signal).await?; + + *size = Some(engine_order); + + let limit_px = round_price(signal.entry_price); + let sz = round_price(engine_order.size / signal.entry_price); + + let order = BatchOrder { + orders: vec![ + OrderRequest { + asset: asset_id, + is_buy: matches!(signal.side, Side::Bid), + limit_px, + sz, + reduce_only: false, + order_type: OrderTypePlacement::Limit { + tif: TimeInForce::Gtc, }, - OrderRequest { - asset: asset_id, - 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, - tpsl: hypercore::TpSl::Tp, - }, - cloid: Default::default(), + cloid: engine_order.entry, + }, + OrderRequest { + asset: asset_id, + is_buy: !matches!(signal.side, Side::Bid), + limit_px, + sz, + reduce_only: true, + order_type: OrderTypePlacement::Trigger { + is_market: true, + trigger_px: round_price(signal.take_profit), + tpsl: hypercore::TpSl::Tp, }, - OrderRequest { - asset: asset_id, - 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, - tpsl: hypercore::TpSl::Sl, - }, - cloid: Default::default(), + cloid: engine_order.take_profit, + }, + OrderRequest { + asset: asset_id, + is_buy: !matches!(signal.side, Side::Bid), + limit_px, + sz, + reduce_only: true, + order_type: OrderTypePlacement::Trigger { + is_market: true, + trigger_px: round_price(engine_order.stop_loss), + tpsl: hypercore::TpSl::Sl, }, - ], - grouping: hypercore::OrderGrouping::Na, - builder: None, - }; + cloid: engine_order.stop_loss_id, + }, + ], + grouping: hypercore::OrderGrouping::NormalTpsl, + builder: None, + }; - let nonce = chrono::Utc::now().timestamp_millis() as u64; + println!("{order:?}"); - 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?; - } + let nonce = NonceHandler::default(); + + match client + .place(&acc.private_key.0, order, nonce.next(), None, None) + .await + { + // Invalid order + Ok(o) if o.iter().any(|o| matches!(o, OrderResponseStatus::Error(_))) => { + return Err(anyhow::anyhow!( + "order rejected by HyperLiquid: {}", + o.into_iter() + .enumerate() + .filter_map(|(i, res)| { + if let OrderResponseStatus::Error(e) = res { + Some(format!("{i}={e}")) + } else { + None + } + }) + .collect::>() + .join(", ") + )); + } + + Ok(_) => { + self.risk_engine.order_placed(engine_order).await; + } + + e => { + e?; } - } 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 492dc7f..fe2fd7f 100644 --- a/src/engine/engine/mod.rs +++ b/src/engine/engine/mod.rs @@ -1,16 +1,17 @@ pub mod command; pub mod execution; +pub mod risk; pub mod strategy; pub mod terminal; use crate::{ - engine::{strategy::StrategyEngine, terminal::TerminalServer}, - store::{accounts::AccountList, config::Config}, + engine::{risk::RiskEngine, strategy::StrategyEngine, terminal::TerminalServer}, + store::{accounts::AccountList, config::ConfigManager}, }; use hypersdk::hypercore::ws::ConnectionStream; use pulse_sdk::prelude::*; -use std::{collections::HashMap, sync::Arc}; -use tokio::sync::Mutex; +use std::{collections::HashMap, sync::Arc, time::Duration}; +use tokio::{sync::Mutex, time::Instant}; #[derive(Debug, Clone)] pub struct WatchList { @@ -23,10 +24,10 @@ pub struct Engine { // engine pub terminal_server: Arc, pub strategy_engine: Arc, - pub ws_stream: Arc>, + pub risk_engine: Arc, // data - pub config: Arc>, + pub config: Arc>, pub accounts: Arc>, // live data / status @@ -35,38 +36,52 @@ pub struct Engine { pub signals: Arc>>, } +pub struct EngineSides { + pub strategy_stream: ConnectionStream, + pub risk_stream: ConnectionStream, +} + impl Engine { - pub async fn new() -> tokio::io::Result> { - let config = Config::new().await?; + pub async fn new() -> tokio::io::Result<(Arc, EngineSides)> { + let config = ConfigManager::new().await?; - let (ws_handle, ws_stream) = hypersdk::hypercore::mainnet_ws().split(); + let (strategy_handle, strategy_stream) = hypersdk::hypercore::mainnet_ws().split(); + let (risk_handle, risk_stream) = hypersdk::hypercore::mainnet_ws().split(); - let strategy = StrategyEngine::new(&config.strategy, ws_handle).await?; + let strategy = StrategyEngine::new(&config.get_ref()?.strategy, strategy_handle).await?; let accounts = Arc::new(Mutex::new(AccountList::new().await?)); let config = Arc::new(Mutex::new(config)); - Ok(Arc::new_cyclic(|engine| Self { - // engine - terminal_server: TerminalServer::new(engine.clone()), - strategy_engine: strategy.initialize(engine.clone()), - ws_stream: Arc::new(Mutex::new(ws_stream)), + Ok(( + Arc::new_cyclic(|engine| Self { + // engine + terminal_server: TerminalServer::new(engine.clone()), + strategy_engine: strategy.initialize(engine.clone()), + risk_engine: RiskEngine::new(engine.clone(), risk_handle), - // data - config, - accounts, + // data + config, + accounts, - // live data / status - status: Arc::new(Mutex::new(EngineStatus { - strategy_mode: Mode::Auto, - strategy_state: ItemState::Stopped, - })), - watch_list: Arc::new(Mutex::new(WatchList { - name_to_index: HashMap::new(), - items: Vec::new(), - })), - signals: Arc::new(Mutex::new(Vec::new())), - })) + // live data / status + status: Arc::new(Mutex::new(EngineStatus { + perps_equity: 0.into(), + strategy_mode: Mode::Auto, + strategy_state: ItemState::Stopped, + positions: Vec::new(), + })), + watch_list: Arc::new(Mutex::new(WatchList { + name_to_index: HashMap::new(), + items: Vec::new(), + })), + signals: Arc::new(Mutex::new(Vec::new())), + }), + EngineSides { + strategy_stream, + risk_stream, + }, + )) } /// Starts the main strategy server @@ -89,12 +104,53 @@ impl Engine { } } - pub async fn get_config_status(&self) -> EngineConfig { - EngineConfig { - strategy: self.strategy_engine.strategy.lock().await.manifest.clone(), + pub async fn run_daily_scheduler(self: Arc) -> anyhow::Result<()> { + loop { + self.risk_engine.day_tick().await?; + + let now_local = chrono::Local::now(); + + let midnight = chrono::NaiveTime::from_hms_opt(0, 0, 0).unwrap(); + + let tomorrow_local = now_local + .date_naive() + .succ_opt() + .unwrap() + .and_time(midnight) + .and_local_timezone(chrono::Local) + .unwrap(); + + let duration_until_midnight = tomorrow_local.signed_duration_since(now_local); + + let std_duration = Duration::from_secs(duration_until_midnight.num_seconds() as u64); + + let deadline = Instant::now() + std_duration; + + self.terminal_server + .info( + "engine::schedule", + &format!("next daily reset: {tomorrow_local}"), + ) + .await?; + + tokio::time::sleep_until(deadline).await; } } + pub async fn get_config_status(&self) -> tokio::io::Result { + let manager = self.config.lock().await; + let config = manager.get_ref()?; + + Ok(EngineConfig { + preset: manager.preset.clone(), + strategy: self.strategy_engine.strategy.lock().await.manifest.clone(), + description: config.description.clone(), + + risk_per_trade: config.risk.risk_per_trade.to_string(), + max_daily_loss: config.risk.max_daily_loss.to_string(), + }) + } + pub async fn update_status(&self) -> tokio::io::Result<()> { self.terminal_server .broadcast(TerminalServerMessage::StatusUpdated( diff --git a/src/engine/engine/risk.rs b/src/engine/engine/risk.rs new file mode 100644 index 0000000..28569fc --- /dev/null +++ b/src/engine/engine/risk.rs @@ -0,0 +1,182 @@ +use pulse_sdk::prelude::*; + +use futures::StreamExt; +use hypersdk::{ + Decimal, + hypercore::{ + self, Cloid, Side, + ws::{ConnectionHandle, ConnectionStream, Event}, + }, +}; +use std::sync::{Arc, Weak}; +use tokio::sync::Mutex; + +use crate::{engine::Engine, store::config::RiskConfig}; + +pub struct RiskState { + pub starting_equity: Decimal, + pub pnl: Decimal, +} + +pub struct RiskEngine { + // Orders made by the engine + pub orders: Mutex>, + pub handle: Mutex, + pub state: Mutex, + pub engine: Weak, +} + +impl RiskEngine { + pub fn new(engine: Weak, handle: ConnectionHandle) -> Arc { + Arc::new(Self { + orders: Mutex::new(Vec::new()), + handle: Mutex::new(handle), + state: Mutex::new(RiskState { + starting_equity: 0.into(), + pnl: 0.into(), + }), + engine, + }) + } + + pub async fn day_tick(&self) -> anyhow::Result<()> { + let engine = self.get_engine(); + + let client = hypercore::mainnet(); + + let mut state = self.state.lock().await; + + if let Some(acc) = engine.accounts.lock().await.get_active() { + let clearing_house = client.clearinghouse_state(acc.address, None).await?; + + state.starting_equity = clearing_house.margin_summary.account_value; + } + + state.pnl = 0.into(); + + Ok(()) + } + + pub async fn run_event_stream( + self: Arc, + mut stream: ConnectionStream, + ) -> anyhow::Result<()> { + let engine = self.get_engine(); + + while let Some(e) = stream.next().await { + match e { + Event::Message(hypercore::Incoming::OrderUpdates(order_updates)) => { + for order in order_updates { + if let Some(cloid) = order.order.cloid { + let orders = self.orders.lock().await.clone(); + + let mut rm = Vec::new(); + + for (i, order_ids) in orders.iter().enumerate() { + if order_ids.stop_loss_id == cloid { + if order.status.is_filled() { + rm.push(i); + } + + break; + } + } + + for r in rm { + let order = self.orders.lock().await.remove(r); + + self.state.lock().await.pnl -= order.risk; + } + } + } + } + + Event::Connected => { + engine + .terminal_server + .info("engine::risk", "HyperLiquid WebSocket connected") + .await?; + } + + Event::Disconnected => { + engine + .terminal_server + .warn("engine::risk", "HyperLiquid WebSocket disconnected") + .await?; + } + + _ => { + engine + .terminal_server + .warn( + "engine::risk", + &format!("Unexpected event from HyperLiquid WebSocket: {e:?}"), + ) + .await?; + } + } + } + + Ok(()) + } + + pub async fn validate_signal(&self, _signal: &mut Signal) -> tokio::io::Result { + let state = self.state.lock().await; + + let under_max_losses = state.pnl + < self + .get_risk_config() + .await? + .max_daily_loss + .get(state.starting_equity); + + Ok(under_max_losses) + } + + pub async fn create_order(&self, signal: &Signal) -> tokio::io::Result { + let state = self.state.lock().await; + + let risk = self + .get_risk_config() + .await? + .risk_per_trade + .get(state.starting_equity); + + let size = self + .get_risk_config() + .await? + .size_per_trade + .get(state.starting_equity); + + let sl_distance = risk / (size / signal.entry_price); + + Ok(EngineOrder { + entry: Cloid::random(), + take_profit: Cloid::random(), + stop_loss_id: Cloid::random(), + + size, + risk, + + stop_loss: if matches!(signal.side, Side::Ask) { + signal.entry_price + sl_distance + } else { + signal.entry_price - sl_distance + }, + }) + } + + pub async fn order_placed(&self, order: EngineOrder) { + self.orders.lock().await.push(order); + } + + pub async fn get_risk_config(&self) -> tokio::io::Result { + Ok(self.get_engine().config.lock().await.get_ref()?.risk) + } + + pub fn get_engine(&self) -> Arc { + self.engine + .upgrade() + .expect("Failed to upgrade engine(Weak) to Arc") + } +} diff --git a/src/engine/engine/strategy.rs b/src/engine/engine/strategy.rs index b79e26e..d407b60 100644 --- a/src/engine/engine/strategy.rs +++ b/src/engine/engine/strategy.rs @@ -1,6 +1,11 @@ -use anyhow::Context; -use hypersdk::hypercore::{self, CandleInterval, Subscription, ws::ConnectionHandle}; use pulse_sdk::prelude::*; + +use anyhow::Context; +use futures::StreamExt; +use hypersdk::hypercore::{ + self, CandleInterval, Subscription, + ws::{ConnectionHandle, ConnectionStream, Event}, +}; use std::{ collections::HashSet, sync::{Arc, Weak}, @@ -75,11 +80,40 @@ impl StrategyEngine { Ok(()) } + pub async fn run_event_stream( + self: Arc, + mut stream: ConnectionStream, + ) -> anyhow::Result<()> { + let engine = self.get_engine(); + + while let Some(e) = stream.next().await { + match e { + Event::Message(incoming) => { + self.send(&StrategyEngineMessage::Incoming(incoming)) + .await?; + } + + Event::Connected => { + engine + .terminal_server + .info("engine::strategy", "HyperLiquid WebSocket connected") + .await?; + } + + Event::Disconnected => { + engine + .terminal_server + .warn("engine::strategy", "HyperLiquid WebSocket disconnected") + .await?; + } + } + } + + Ok(()) + } + pub async fn run(self: &Arc) -> anyhow::Result<()> { - let engine = self - .engine - .upgrade() - .expect("Failed to upgrade engine (StrategyEngine)"); + let engine = self.get_engine(); let mut stdout = { let mut child = self.strategy.lock().await; @@ -117,18 +151,33 @@ impl StrategyEngine { engine.terminal_server.log_raw(log).await?; } - Some(StrategyMessage::Signal(signal)) => { - match engine.execute_signal(&signal).await { - Ok(_) => engine.signals.lock().await.push(Ok(signal)), - Err(e) => { - engine.signals.lock().await.push(Err(signal)); + Some(StrategyMessage::Signal(mut signal)) => { + if !engine.risk_engine.validate_signal(&mut signal).await? { + engine.signals.lock().await.push(Err((signal, None))); + engine + .terminal_server + .error("engine::risk", "Signal rejected") + .await?; + + continue; + } + + let mut size = None; + + let result = match engine.execute_signal(&mut size, &signal).await { + Ok(_) => Ok((signal, size)), + Err(e) => { engine .terminal_server - .error("signal", &format!("Failed to execute signal: {e}")) - .await? + .error("engine::order", &format!("Failed to execute signal: {e}")) + .await?; + + Err((signal, size)) } - } + }; + + engine.signals.lock().await.push(result); engine .terminal_server @@ -197,4 +246,10 @@ impl StrategyEngine { } } } + + pub fn get_engine(&self) -> Arc { + self.engine + .upgrade() + .expect("Failed to upgrade engine (StrategyEngine)") + } } diff --git a/src/engine/engine/terminal.rs b/src/engine/engine/terminal.rs index f2e6211..781ac9b 100644 --- a/src/engine/engine/terminal.rs +++ b/src/engine/engine/terminal.rs @@ -1,5 +1,6 @@ -use crate::engine::Engine; use pulse_sdk::{map_postcard_err, prelude::*}; + +use crate::engine::Engine; use std::{ collections::HashMap, sync::{Arc, Weak}, @@ -37,7 +38,7 @@ impl TerminalServer { } pub async fn run_server(self: Arc) -> tokio::io::Result<()> { - let path = pulse_sdk::server_path(); + let path = server_path(); if path.exists() { tokio::fs::remove_file(&path).await?; @@ -91,7 +92,7 @@ impl TerminalServer { self.send_to( id, - TerminalServerMessage::ConfigUpdated(engine.get_config_status().await), + TerminalServerMessage::ConfigUpdated(engine.get_config_status().await?), ) .await?; @@ -103,7 +104,13 @@ impl TerminalServer { self.send_to( id, - pulse_sdk::terminal::TerminalServerMessage::SetLogs(self.logs.lock().await.clone()), + TerminalServerMessage::SetLogs(self.logs.lock().await.clone()), + ) + .await?; + + self.send_to( + id, + TerminalServerMessage::SignalsUpdated(engine.signals.lock().await.clone()), ) .await?; @@ -151,7 +158,7 @@ impl TerminalServer { pub async fn broadcast( self: &Arc, - message: pulse_sdk::terminal::TerminalServerMessage, + message: TerminalServerMessage, ) -> tokio::io::Result<()> { let msg = map_postcard_err(postcard::to_allocvec(&message))?; @@ -194,8 +201,11 @@ impl TerminalServer { loop { refresh.tick().await; - match crate::fetch::fetch_watch_list(&client, &engine.config.lock().await.watchlist) - .await + match crate::fetch::fetch_watch_list( + &client, + &engine.config.lock().await.get_ref()?.watchlist, + ) + .await { Ok(watch_list) => { *engine.watch_list.lock().await = watch_list.clone(); @@ -224,19 +234,23 @@ impl TerminalServer { 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?; + let mut status = engine.status.lock().await; + + status.perps_equity = state.margin_summary.account_value; + status.positions = 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(); + + drop(status); + + engine.update_status().await?; } Err(e) => { self.error("orders", &format!("Unable to get open orders: {e}")) @@ -256,7 +270,7 @@ impl TerminalServer { pub async fn send_to( self: &Arc, id: &usize, - message: pulse_sdk::terminal::TerminalServerMessage, + message: TerminalServerMessage, ) -> tokio::io::Result<()> { Self::send_to_client( self.clients.lock().await.get_mut(id).ok_or_else(|| { @@ -312,8 +326,7 @@ impl TerminalServer { self.logs.lock().await.push(log.clone()); - self.broadcast(pulse_sdk::terminal::TerminalServerMessage::AddLog(log)) - .await + self.broadcast(TerminalServerMessage::AddLog(log)).await } pub async fn info(self: &Arc, name: &str, message: &str) -> tokio::io::Result<()> { diff --git a/src/engine/main.rs b/src/engine/main.rs index c9d3244..fa87e1b 100644 --- a/src/engine/main.rs +++ b/src/engine/main.rs @@ -4,15 +4,36 @@ pub mod store; #[tokio::main] async fn main() -> anyhow::Result<()> { - let engine = engine::Engine::new().await?; + let (engine, sides) = engine::Engine::new().await?; let server = engine.terminal_server.spawn_server().await; let broadcaster = engine.terminal_server.spawn_broadcaster().await; + let risk = tokio::spawn( + engine + .risk_engine + .clone() + .run_event_stream(sides.risk_stream), + ); + + let strategy = tokio::spawn( + engine + .strategy_engine + .clone() + .run_event_stream(sides.strategy_stream), + ); + + let daily_scheduler = tokio::spawn(engine.clone().run_daily_scheduler()); + engine.run().await?; - server.await??; - broadcaster.await??; + server.abort(); + broadcaster.abort(); + + risk.abort(); + strategy.abort(); + + daily_scheduler.abort(); Ok(()) } diff --git a/src/engine/store/config.rs b/src/engine/store/config.rs index ba42aa2..d3c6f03 100644 --- a/src/engine/store/config.rs +++ b/src/engine/store/config.rs @@ -1,23 +1,81 @@ -use hypersdk::hypercore::CandleInterval; +use pulse_sdk::general::Allocation; + +use std::collections::HashMap; + +#[derive(Debug, Clone, Copy, serde::Serialize, serde::Deserialize)] +pub struct RiskConfig { + pub size_per_trade: Allocation, + pub risk_per_trade: Allocation, + pub max_daily_loss: Allocation, +} #[derive(Debug, serde::Serialize, serde::Deserialize)] pub struct Config { + pub description: Option, pub watchlist: Vec, pub strategy: String, - pub cooldown: CandleInterval, + pub risk: RiskConfig, +} + +#[derive(Debug, serde::Serialize, serde::Deserialize)] +pub struct ConfigManager { + pub preset: String, + + #[serde(flatten)] + pub configs: HashMap, +} + +impl Default for RiskConfig { + fn default() -> Self { + Self { + size_per_trade: Allocation::Percent(20.into()), + risk_per_trade: Allocation::Percent(10.into()), + max_daily_loss: Allocation::Percent(20.into()), + } + } } impl Default for Config { fn default() -> Self { Self { + description: None, watchlist: vec!["BTC".to_string(), "SOL".to_string(), "ETH".to_string()], strategy: String::new(), - cooldown: CandleInterval::ThirtyMinutes, + risk: Default::default(), } } } -impl Config { +impl Default for ConfigManager { + fn default() -> Self { + Self { + preset: "Main".to_string(), + configs: HashMap::from([("Main".to_string(), Config::default())]), + } + } +} + +impl ConfigManager { + pub fn get_ref(&self) -> tokio::io::Result<&Config> { + self.configs.get(&self.preset).ok_or_else(|| { + tokio::io::Error::new( + std::io::ErrorKind::NotFound, + format!("Failed to get preset '{}'", self.preset), + ) + }) + } + + pub fn get_mut(&mut self) -> tokio::io::Result<&mut Config> { + self.configs.get_mut(&self.preset).ok_or_else(|| { + tokio::io::Error::new( + std::io::ErrorKind::NotFound, + format!("Failed to get preset '{}'", self.preset), + ) + }) + } +} + +impl ConfigManager { pub async fn new() -> tokio::io::Result { let path = crate::store::pulse_config_file()?; diff --git a/src/engine/store/strategy.rs b/src/engine/store/strategy.rs index 858fd9a..bfcb8cd 100644 --- a/src/engine/store/strategy.rs +++ b/src/engine/store/strategy.rs @@ -1,9 +1,6 @@ -use std::{path::PathBuf, process::Stdio}; +use pulse_sdk::{map_postcard_err, prelude::*}; -use pulse_sdk::{ - map_postcard_err, - strategy::{StrategyEngineMessage, StrategyManifest, StrategyMessage}, -}; +use std::{path::PathBuf, process::Stdio}; use tokio::{ io::{AsyncReadExt, AsyncWriteExt}, process::{Child, ChildStdout, Command}, diff --git a/src/terminal/formatting.rs b/src/terminal/formatting.rs index 64744a6..4b3db06 100644 --- a/src/terminal/formatting.rs +++ b/src/terminal/formatting.rs @@ -1,6 +1,7 @@ -use hypersdk::hypercore::Side; use pulse_sdk::prelude::*; +use hypersdk::hypercore::Side; + pub trait Formatted { fn get_formatted(&self) -> Vec; } @@ -54,32 +55,28 @@ impl Formatted for EventLog { impl Formatted for SignalStatus { fn get_formatted(&self) -> Vec { match self { - Ok(signal) => { - vec![ - format!("\x1b[33mOK\x1b[0m"), - if matches!(signal.side, Side::Ask) { + Ok((signal, order)) | Err((signal, order)) => { + let mut base = vec![ + if matches!(self, Ok(_)) { + format!("\x1b[34mOK\x1b[0m") + } else { + format!("\x1b[31mREJ\x1b[0m") + }, + if matches!(signal.side, Side::Bid) { format!("\x1b[32mBUY\x1b[0m") } else { format!("\x1b[31mSELL\x1b[0m") }, format_symbol(&signal.symbol), format_usd(signal.entry_price.as_f64()), - format!("\x1b[33mAPR {}\x1b[0m", signal.confidence), - ] - } + format_usd(signal.take_profit.as_f64()), + ]; - Err(signal) => { - vec![ - format!("\x1b[31mERR\x1b[0m"), - if matches!(signal.side, Side::Ask) { - format!("\x1b[32mBUY\x1b[0m") - } else { - format!("\x1b[31mSELL\x1b[0m") - }, - format_symbol(&signal.symbol), - format_usd(signal.entry_price.as_f64()), - format!("\x1b[33mAPR {}\x1b[0m", signal.confidence), - ] + if let Some(order) = order { + base.extend(vec![format_usd_reverse(order.stop_loss.as_f64())]); + } + + base } } } @@ -199,14 +196,14 @@ impl Formatted for EngineStatus { fn get_formatted(&self) -> Vec { vec![ Triple( - "\x1b[2mState\x1b[0m", - "\x1b[2mMode\x1b[0m", - "\x1b[2m..\x1b[0m", + "\x1b[2mPerps Equity\x1b[0m", + "\x1b[2mStrategy State\x1b[0m", + "\x1b[2mStrategy Mode\x1b[0m", ), Triple( + &format_usd(self.perps_equity.as_f64()), &self.strategy_state.to_string(), &self.strategy_mode.to_string(), - "..", ), ] .get_formatted() @@ -217,11 +214,25 @@ impl Formatted for EngineConfig { fn get_formatted(&self) -> Vec { vec![ Triple( + "\x1b[2mPreset\x1b[0m", "\x1b[2mStrategy\x1b[0m", - "\x1b[2m..\x1b[0m", - "\x1b[2m..\x1b[0m", + "\x1b[2mDescription\x1b[0m", ), - Triple(&self.strategy.name, "..", ".."), + Triple( + &self.preset, + &self.strategy.name, + self.description + .as_ref() + .map(String::as_str) + .unwrap_or("none"), + ), + Triple("", "", ""), + Triple( + "\x1b[2mRisk/Trade\x1b[0m", + "\x1b[2m..\x1b[0m", + "\x1b[2mMax daily loss\x1b[0m", + ), + Triple(&self.risk_per_trade, "..", &self.max_daily_loss), ] .get_formatted() } @@ -256,6 +267,14 @@ pub fn format_usd(value: f64) -> String { } } +pub fn format_usd_reverse(value: f64) -> String { + if value.is_sign_positive() { + format!("\x1b[31m${}\x1b[0m", format_f64(value)) + } else { + format!("\x1b[32m${}\x1b[0m", format_f64(value)) + } +} + pub fn format_f64(value: f64) -> String { let abs = value.abs(); diff --git a/src/terminal/main.rs b/src/terminal/main.rs index dc8cb13..5b80956 100644 --- a/src/terminal/main.rs +++ b/src/terminal/main.rs @@ -2,6 +2,8 @@ pub mod command; pub mod formatting; pub mod terminal; +use pulse_sdk::prelude::*; + use std::any::Any; use chrono::{Local, Utc}; @@ -21,8 +23,6 @@ use pulse_ui::{ use crate::formatting::{Formatted, apply_padding}; -use pulse_sdk::prelude::*; - pub struct PulseTradeApp { sock: Option, @@ -33,7 +33,6 @@ pub struct PulseTradeApp { status: State>, watch_list: State>, - active_positions: State>, logs: State>, signals: State>, inspect: State, @@ -128,7 +127,10 @@ impl App for PulseTradeApp { ( LayoutItem::Widget(Size::Flex(1)), Box::new( - advanced_draw(&self.scroll, 1, "POSITIONS", &self.active_positions).await, + advanced_draw_map(&self.scroll, 1, "POSITIONS", &self.status, |status| { + &status.positions + }) + .await, ), ), ( @@ -177,7 +179,6 @@ async fn main() -> tokio::io::Result<()> { command: ctx.use_state(InputState::new()), scroll: ctx.use_state(ScrollState(1, [0; 7])), watch_list: ctx.use_state(Vec::new()), - active_positions: ctx.use_state(Vec::new()), signals: ctx.use_state(Vec::new()), logs: ctx.use_state(Vec::new()), inspect: ctx.use_state(InspectTarget::None), @@ -190,6 +191,26 @@ async fn main() -> tokio::io::Result<()> { Ok(()) } +pub async fn advanced_draw_map( + scroll: &State>, + index: usize, + title: &'static str, + state: &State>, + m: impl for<'a> FnOnce(&'a T) -> &'a B, +) -> ScrollText { + let scroll = scroll.lock().await; + + scroll.scroll( + index, + format!(" {}{title}\x1b[0m", scroll.get_selected(index)), + if let Some(state) = &*state.lock().await { + apply_padding(m(state).get_formatted()).join("\n") + } else { + " Loading..".to_string() + }, + ) +} + pub async fn advanced_draw( scroll: &State>, index: usize, diff --git a/src/terminal/terminal.rs b/src/terminal/terminal.rs index 9acf585..ed503b7 100644 --- a/src/terminal/terminal.rs +++ b/src/terminal/terminal.rs @@ -1,4 +1,5 @@ use pulse_sdk::{map_postcard_err, prelude::*}; + use pulse_ui::state::State; use tokio::{ @@ -46,7 +47,6 @@ impl TerminalClient { let reader = reader.expect("Reader failed to swap"); let watch_list = app.watch_list.clone(); - let active_positions = app.active_positions.clone(); let logs = app.logs.clone(); let signals = app.signals.clone(); let market_overview = app.config.clone(); @@ -57,7 +57,6 @@ impl TerminalClient { if let Err(v) = Self::run_client( reader, watch_list, - active_positions, logs, signals, market_overview, @@ -66,7 +65,27 @@ impl TerminalClient { ) .await { - panic!("{v}") + crossterm::terminal::disable_raw_mode().unwrap(); + + crossterm::execute!(std::io::stdout(), crossterm::cursor::Show).unwrap(); + + crossterm::execute!( + std::io::stdout(), + crossterm::terminal::Clear(crossterm::terminal::ClearType::Purge) + ) + .unwrap(); + + crossterm::execute!( + std::io::stdout(), + crossterm::terminal::Clear(crossterm::terminal::ClearType::All) + ) + .unwrap(); + + crossterm::execute!(std::io::stdout(), crossterm::cursor::MoveTo(0, 0)).unwrap(); + + println!("Error using app: {v}"); + + std::process::exit(1); } }); @@ -79,7 +98,6 @@ impl TerminalClient { mut reader: OwnedReadHalf, watch_list: State>, - active_positions: State>, logs: State>, signals: State>, config: State>, @@ -107,10 +125,6 @@ impl TerminalClient { *watch_list.lock().await = v; } - TerminalServerMessage::PositionsUpdated(v) => { - *active_positions.lock().await = v; - } - TerminalServerMessage::ConfigUpdated(v) => { *config.lock().await = Some(v); }