Merge pull request #17 from selimaj-dev/orders-and-strategy-redesign

Orders and strategy redesign

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",

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" }

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 }

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,

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<T>(res: postcard::Result<T>) -> tokio::io::Result<T> {
     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::<usize>()];
+                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<String>) -> tokio::io::Result<()> {
+        unimplemented!("Strategy::command")
+    }
+
+    async fn watchlist(&self, _watchlist: Vec<prelude::MarketItem>) -> 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<hypersdk::hypercore::Candle>,
+    ) -> tokio::io::Result<()> {
+        unimplemented!("Strategy::candlestick")
+    }
+}

pulse-sdk/src/plugin.rs

@@ -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<MarketItem>),
-
-    Command {
-        command: String,
-        args: Vec<String>,
-    },
-
-    CandleUpdate {
-        symbol: String,
-        interval: CandleInterval,
-        candle: Candle,
-    },
-
-    Candlestick {
-        symbol: String,
-        interval: CandleInterval,
-        candles: Vec<Candle>,
-    },
-}
-
-#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
-pub enum RiskEngineMessage {
-    Initialize,
-
-    WatchList(Vec<MarketItem>),
-
-    Command { command: String, args: Vec<String> },
-
-    Signal(StrategySignal),
-}
-
-#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
-pub struct StrategySignal {
-    pub symbol: String,
-    pub side: Direction,
-    pub confidence: f32,
-    pub price: Option<f64>,
-}

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<MarketItem>),
+
+    Command {
+        command: String,
+        args: Vec<String>,
+    },
+
+    Incoming(Incoming),
+
+    Candlestick {
+        symbol: String,
+        interval: CandleInterval,
+        candles: Vec<Candle>,
+    },
+}

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<Signal, Signal>;
 
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
 pub enum TerminalServerMessage {
@@ -17,7 +19,7 @@ pub enum TerminalServerMessage {
     StrategyUpdated(Strategy),
 
     // Signals
-    SignalsUpdated(Vec<Signal>),
+    SignalsUpdated(Vec<SignalStatus>),
 
     // 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,

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()))
         }
     }
 }

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?;
                                 }
                             }

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(())
+    }
+}

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<String, usize>,
+    pub items: Vec<MarketItem>,
+}
+
 #[derive(Clone)]
 pub struct Engine {
     pub terminal_server: Arc<TerminalServer>,
     pub strategy: Arc<StrategyEngine>,
     pub config: Arc<Mutex<Config>>,
     pub accounts: Arc<Mutex<AccountList>>,
-    pub watch_list: Arc<Mutex<Vec<MarketItem>>>,
+    pub watch_list: Arc<Mutex<WatchList>>,
+    pub signals: Arc<Mutex<Vec<SignalStatus>>>,
 }
 
 impl Engine {
     pub async fn new() -> tokio::io::Result<Arc<Self>> {
         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,8 +99,7 @@ impl Engine {
                 match client.clearinghouse_state(acc.address, None).await {
                     Ok(state) => {
                         self.terminal_server
-                            .broadcast(
-                                pulse_sdk::terminal::TerminalServerMessage::PositionsUpdated(
+                            .broadcast(TerminalServerMessage::PositionsUpdated(
                                 state
                                     .asset_positions
                                     .into_iter()
@@ -102,13 +109,11 @@ impl Engine {
                                         entry_price: USD(position
                                             .position
                                             .entry_px
-                                                .map(|px| px.as_f64())
-                                                .unwrap_or(0.0)),
-                                            profit: USD(position.position.unrealized_pnl.as_f64()),
+                                            .unwrap_or_default()),
+                                        profit: USD(position.position.unrealized_pnl),
                                     })
                                     .collect(),
-                                ),
-                            )
+                            ))
                             .await?;
                     }
                     Err(e) => {

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<Plugin<StrategyEngineMessage, StrategyMessage, StrategyManifest>>,
-    pub risk: Arc<Plugin<RiskEngineMessage, RiskMessage, RiskManifest>>,
+    pub strategy: Arc<StrategyChild>,
     pub engine: Weak<Engine>,
 
     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<Self> {
-        let strategy = pulse_plugin(strategy_id)?;
-        let risk = pulse_plugin(risk_id)?;
+    pub async fn new(strategy_id: &str) -> tokio::io::Result<Self> {
+        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<Self>) {
         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<Self>, 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<Self>, 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())

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<f64, String> {
+use crate::engine::WatchList;
+
+fn number(value: &Value, field: &str) -> Result<Decimal, String> {
     let raw = value[field]
         .as_str()
         .ok_or_else(|| format!("asset context field {field} must be a string"))?;
 
-    raw.parse::<f64>()
+    raw.parse::<Decimal>()
         .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<Vec<MarketItem>, String> {
+) -> Result<WatchList, String> {
     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()),
+        items.push(MarketItem {
+            symbol: Symbol(symbol.clone()),
             price: USD(price),
             volume_24h: USD(volume_24h),
-                trend: ((price / previous_day_price) - 1.0) * 100.0,
-            },
-        );
+            trend: ((price / previous_day_price) - <Decimal as From<i32>>::from(1))
+                * <Decimal as From<i32>>::from(100),
+        });
     }
 
-    symbols
-        .iter()
-        .map(|symbol| {
-            by_symbol
-                .remove(symbol.as_str())
-                .ok_or_else(|| format!("{symbol} is not in the Hyperliquid perpetual universe"))
+    Ok(WatchList {
+        items,
+        name_to_index,
     })
-        .collect()
 }

src/engine/store/config.rs

@@ -4,7 +4,6 @@ use hypersdk::hypercore::CandleInterval;
 pub struct Config {
     pub watchlist: Vec<String>,
     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,
         }
     }

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<PathBuf> {
     std::env::home_dir().ok_or_else(|| {
@@ -14,12 +14,12 @@ pub fn pulse_directory() -> tokio::io::Result<PathBuf> {
     Ok(home_dir()?.join(".config").join("pulse-trader"))
 }
 
-pub fn pulse_plugins_directory() -> tokio::io::Result<PathBuf> {
-    Ok(pulse_directory()?.join("plugins"))
+pub fn pulse_strategies_directory() -> tokio::io::Result<PathBuf> {
+    Ok(pulse_directory()?.join("strategies"))
 }
 
-pub fn pulse_plugin(id: &str) -> tokio::io::Result<PathBuf> {
-    Ok(pulse_directory()?.join("plugins").join(id))
+pub fn pulse_strategy(id: &str) -> tokio::io::Result<PathBuf> {
+    Ok(pulse_directory()?.join("strategies").join(id))
 }
 
 pub fn pulse_config_file() -> tokio::io::Result<PathBuf> {

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<S: Serialize, R: for<'de> Deserialize<'de>, M: for<'de> Deserialize<'de>> {
-    pub manifest: Mutex<M>,
+pub struct StrategyChild {
+    pub manifest: Mutex<StrategyManifest>,
     pub stdout: Mutex<ChildStdout>,
     pub process: Mutex<Child>,
-
-    pub _p: (PhantomData<S>, PhantomData<R>),
 }
 
-impl<S: Serialize, R: for<'de> Deserialize<'de>, M: for<'de> Deserialize<'de>> Plugin<S, R, M> {
-    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<Option<R>> {
+    pub async fn recv(&self) -> tokio::io::Result<Option<StrategyMessage>> {
         let mut stdout = self.stdout.lock().await;
 
         let mut len_buf = [0u8; size_of::<usize>()];
@@ -46,7 +42,7 @@ impl<S: Serialize, R: for<'de> 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<S: Serialize, R: for<'de> 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?;

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(())

src/terminal/formatting.rs

@@ -58,18 +58,38 @@ impl Formatted for EventLog {
     }
 }
 
-impl Formatted for Signal {
+impl Formatted for SignalStatus {
     fn get_formatted(&self) -> Vec<String> {
+        match self {
+            Ok(signal) => {
                 vec![
-            if matches!(self.kind, Direction::Buy) {
-                format!("\x1b[32m{}\x1b[0m", self.kind)
+                    format!("\x1b[33mOK\x1b[0m"),
+                    if matches!(signal.kind, Direction::Buy) {
+                        format!("\x1b[32mBUY\x1b[0m")
                     } else {
-                format!("\x1b[31m{}\x1b[0m", self.kind)
+                        format!("\x1b[31mSELL\x1b[0m")
                     },
-            format!("\x1b[35m{}\x1b[0m", self.symbol),
-            self.price.to_string(),
+                    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),
+                ]
+            }
+        }
+    }
 }
 
 impl Formatted for MarketItem {
@@ -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()

src/terminal/main.rs

@@ -35,7 +35,7 @@ pub struct PulseTradeApp {
     watch_list: State<Vec<MarketItem>>,
     active_positions: State<Vec<Position>>,
     logs: State<Vec<EventLog>>,
-    signals: State<Vec<Signal>>,
+    signals: State<Vec<SignalStatus>>,
     inspect: State<InspectTarget>,
 }
 

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,7 +53,8 @@ impl TerminalClient {
         let status = app.status.clone();
         let inspect = app.inspect.clone();
 
-        tokio::spawn(Self::run_client(
+        tokio::spawn(async {
+            if let Err(v) = Self::run_client(
                 reader,
                 watch_list,
                 active_positions,
@@ -62,7 +63,12 @@ impl TerminalClient {
                 market_overview,
                 status,
                 inspect,
-        ));
+            )
+            .await
+            {
+                panic!("{v}")
+            }
+        });
 
         app.sock = Some(self);
 
@@ -75,11 +81,12 @@ impl TerminalClient {
         watch_list: State<Vec<MarketItem>>,
         active_positions: State<Vec<Position>>,
         logs: State<Vec<EventLog>>,
-        signals: State<Vec<Signal>>,
+        signals: State<Vec<SignalStatus>>,
         market_overview: State<Option<Strategy>>,
         status: State<Option<Status>>,
         inspect: State<InspectTarget>,
     ) -> tokio::io::Result<()> {
+        loop {
             let mut len_buf = [0u8; size_of::<usize>()];
             reader
                 .read_exact(&mut len_buf)
@@ -128,7 +135,6 @@ impl TerminalClient {
                     logs.lock().await.push(v);
                 }
             }
-
-        Ok(())
+        }
     }
 }