Merge pull request #11 from selimaj-dev/engine-market-data

Engine market data

Cargo.toml

@@ -10,6 +10,8 @@ pulse-wire = { workspace = true }
 chrono = "0.4.45"
 tokio = { workspace = true, features = ["rt-multi-thread", "macros", "net", "fs", "io-util", "time"] }
 crossterm = { workspace = true }
+hypersdk = "0.2.14"
+serde_json = "1"
 
 [workspace]
 members = ["pulse-macros", "pulse-ui", "pulse-wire"]

pulse-wire/src/terminal.rs

@@ -36,13 +36,16 @@ pub struct WatchListItem {
     symbol: String,
     price: f64,
     trend: f64,
+    volume_24h: f64,
 }
 
 #[pwp]
 pub struct ActivePosition {
-    symbol: String,
+    pub symbol: String,
-    profit: f64,
+    pub size: f64,
-    amount: f64,
+    pub entry_price: f64,
+    pub mark_price: f64,
+    pub profit: f64,
 }
 
 #[pwp]
@@ -137,10 +140,15 @@ pub struct Alert {
 #[pwp]
 pub enum InspectTarget {
     None,
-    Symbol(WatchListItem),
+    Some(Vec<InspectItem>),
-    Position(ActivePosition),
+}
-    Signal(Signal),
+
-    Alert(Alert),
+#[pwp]
+pub enum InspectItem {
+    String(String),
+    Symbol(String),
+    USD(f64),
+    F64(f64),
 }
 
 impl std::fmt::Display for MarketTrend {

src/engine/engine.rs

@@ -0,0 +1,60 @@
+use crate::terminal::TerminalServer;
+
+const WATCH_LIST_SYMBOLS: &[&str] = &["BTC", "ETH", "SOL", "XRP"];
+
+pub struct Engine {
+    pub terminal_server: TerminalServer,
+}
+
+impl Engine {
+    pub fn new() -> Self {
+        Self {
+            terminal_server: TerminalServer::new(),
+        }
+    }
+
+    pub fn spawn_terminal_server(&self) {
+        let terminal_server = self.terminal_server.clone();
+
+        tokio::spawn(async move {
+            terminal_server
+                .run()
+                .await
+                .expect("Failed to run terminal server");
+        });
+    }
+
+    pub fn spawn_broadcaster(&mut self) {
+        let terminal_server = self.terminal_server.clone();
+
+        tokio::spawn(async move {
+            let mut refresh = tokio::time::interval(tokio::time::Duration::from_secs(5));
+
+            loop {
+                refresh.tick().await;
+
+                match crate::fetch::fetch_watch_list(WATCH_LIST_SYMBOLS).await {
+                    Ok(watch_list) => {
+                        if let Err(error) = terminal_server
+                            .broadcast(
+                                pulse_wire::terminal::TerminalServerMessage::WatchListUpdated(
+                                    watch_list,
+                                ),
+                            )
+                            .await
+                        {
+                            eprintln!("Failed to broadcast Hyperliquid watch list: {error}");
+                        }
+                    }
+                    Err(error) => eprintln!("Failed to refresh Hyperliquid watch list: {error}"),
+                }
+            }
+        });
+    }
+
+    pub async fn run_engine(&mut self) -> tokio::io::Result<()> {
+        loop {
+            tokio::time::sleep(tokio::time::Duration::from_millis(5000)).await;
+        }
+    }
+}

src/engine/fetch.rs

@@ -0,0 +1,80 @@
+use std::collections::HashMap;
+
+use pulse_wire::terminal::WatchListItem;
+use serde_json::Value;
+
+fn number(value: &Value, field: &str) -> Result<f64, String> {
+    let raw = value[field]
+        .as_str()
+        .ok_or_else(|| format!("asset context field {field} must be a string"))?;
+
+    raw.parse::<f64>()
+        .map_err(|error| format!("could not parse asset context field {field} ({raw}): {error}"))
+}
+
+pub async fn fetch_watch_list(symbols: &[&str]) -> Result<Vec<WatchListItem>, String> {
+    let response = hypersdk::hypercore::mainnet()
+        .meta_and_asset_ctxs(None)
+        .await
+        .map_err(|error| format!("Hyperliquid metaAndAssetCtxs request failed: {error}"))?;
+
+    let response = response
+        .as_array()
+        .ok_or("metaAndAssetCtxs response must be an array")?;
+
+    if response.len() != 2 {
+        return Err(format!(
+            "metaAndAssetCtxs response must contain meta and contexts, got {} entries",
+            response.len()
+        ));
+    }
+
+    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")?;
+
+    if universe.len() != contexts.len() {
+        return Err(format!(
+            "Hyperliquid returned {} instruments but {} contexts",
+            universe.len(),
+            contexts.len()
+        ));
+    }
+
+    let mut by_symbol = HashMap::with_capacity(universe.len());
+
+    for (meta, context) in universe.iter().zip(contexts) {
+        let symbol = meta["name"]
+            .as_str()
+            .ok_or("instrument metadata is missing a name")?;
+        let price = number(context, "markPx")?;
+        let previous_day_price = number(context, "prevDayPx")?;
+        let volume_24h = number(context, "dayNtlVlm")?;
+
+        if previous_day_price <= 0.0 {
+            return Err(format!("{symbol} has an invalid previous-day price"));
+        }
+
+        by_symbol.insert(
+            symbol,
+            WatchListItem {
+                symbol: symbol.to_owned(),
+                price,
+                trend: ((price / previous_day_price) - 1.0) * 100.0,
+                volume_24h,
+            },
+        );
+    }
+
+    symbols
+        .iter()
+        .map(|symbol| {
+            by_symbol
+                .remove(*symbol)
+                .ok_or_else(|| format!("{symbol} is not in the Hyperliquid perpetual universe"))
+        })
+        .collect()
+}

src/engine/main.rs

@@ -1,114 +1,16 @@
-use pulse_wire::terminal::{ActivePosition, Signal, WatchListItem};
+pub mod engine;
-
+pub mod fetch;
 pub mod terminal;
 
 #[tokio::main]
 async fn main() -> tokio::io::Result<()> {
-    let terminal_server = terminal::TerminalServer::new();
+    let mut engine = engine::Engine::new();
 
-    {
+    engine.spawn_terminal_server();
-        let terminal_server = terminal_server.clone();
 
-        tokio::spawn(async move {
+    engine.spawn_broadcaster();
-            terminal_server
-                .run()
-                .await
-                .expect("Failed to run terminal server");
-        });
-    }
 
-    loop {
+    engine.run_engine().await?;
-        tokio::time::sleep(tokio::time::Duration::from_millis(1500)).await;
 
-        terminal_server
+    Ok(())
-            .broadcast(
-                pulse_wire::terminal::TerminalServerMessage::WatchListUpdated(vec![
-                    WatchListItem {
-                        symbol: "BTC".to_string(),
-                        price: 118_402.12,
-                        trend: 0.82,
-                    },
-                    WatchListItem {
-                        symbol: "ETH".to_string(),
-                        price: 3_912.48,
-                        trend: -0.41,
-                    },
-                    WatchListItem {
-                        symbol: "SOL".to_string(),
-                        price: 182.91,
-                        trend: 2.18,
-                    },
-                    WatchListItem {
-                        symbol: "XRP".to_string(),
-                        price: 2.84,
-                        trend: 1.22,
-                    },
-                ]),
-            )
-            .await?;
-
-        terminal_server
-            .broadcast(
-                pulse_wire::terminal::TerminalServerMessage::PositionsUpdated(vec![
-                    ActivePosition {
-                        symbol: "BTC".to_string(),
-                        profit: 125.50,
-                        amount: 0.25,
-                    },
-                    ActivePosition {
-                        symbol: "SOL".to_string(),
-                        profit: 84.20,
-                        amount: 5.0,
-                    },
-                    ActivePosition {
-                        symbol: "ETH".to_string(),
-                        profit: -32.75,
-                        amount: 1.0,
-                    },
-                    ActivePosition {
-                        symbol: "XRP".to_string(),
-                        profit: 12.30,
-                        amount: 0.75,
-                    },
-                ]),
-            )
-            .await?;
-
-        terminal_server
-            .broadcast(pulse_wire::terminal::TerminalServerMessage::SignalsUpdated(
-                vec![
-                    Signal {
-                        kind: pulse_wire::terminal::SignalKind::Buy,
-                        symbol: "BTC".to_string(),
-                        param: pulse_wire::terminal::SignalParameter::Lim,
-                        price: 118_800.0,
-                    },
-                    Signal {
-                        kind: pulse_wire::terminal::SignalKind::Buy,
-                        symbol: "BTC".to_string(),
-                        param: pulse_wire::terminal::SignalParameter::Tap,
-                        price: 120_000.0,
-                    },
-                    Signal {
-                        kind: pulse_wire::terminal::SignalKind::Buy,
-                        symbol: "BTC".to_string(),
-                        param: pulse_wire::terminal::SignalParameter::Stl,
-                        price: 118_000.0,
-                    },
-                ],
-            ))
-            .await?;
-
-        terminal_server
-            .broadcast(pulse_wire::terminal::TerminalServerMessage::AddLog(
-                pulse_wire::terminal::EventLog {
-                    kind: pulse_wire::terminal::LogKind::Debug,
-                    name: "Engine".to_string(),
-                    message: "Hello, world!".to_string(),
-                },
-            ))
-            .await?;
-
-        println!("Ok");
-    }
 }

src/engine/terminal.rs

@@ -75,14 +75,18 @@ impl TerminalServer {
                         (command, Vec::new())
                     };
 
-                    self.broadcast(pulse_wire::terminal::TerminalServerMessage::AddLog(
+                    match command {
-                        pulse_wire::terminal::EventLog {
+                        _ => {
-                            kind: pulse_wire::terminal::LogKind::Err,
+                            self.broadcast(pulse_wire::terminal::TerminalServerMessage::AddLog(
-                            name: "Command executor".to_string(),
+                                pulse_wire::terminal::EventLog {
-                            message: format!("Command '{}' not found", command),
+                                    kind: pulse_wire::terminal::LogKind::Err,
-                        },
+                                    name: "Command executor".to_string(),
-                    ))
+                                    message: format!("Command '{}' not found", command),
-                    .await?;
+                                },
+                            ))
+                            .await?;
+                        }
+                    }
                 }
             }
         }

src/terminal/formatting.rs

@@ -1,6 +1,6 @@
 use pulse_wire::terminal::{
-    ActivePosition, Alert, AlertLevel, EventLog, InspectTarget, LogKind, MarketOverview, Signal,
+    ActivePosition, Alert, AlertLevel, EventLog, InspectItem, InspectTarget, LogKind,
-    SignalKind, Status, WatchListItem,
+    MarketOverview, Signal, SignalKind, Status, WatchListItem,
 };
 
 pub trait Formatted {
@@ -12,36 +12,15 @@ impl Formatted for InspectTarget {
         match self {
             Self::None => vec!["\x1b[2mnothing to inspect\x1b[0m".to_string()],
 
-            Self::Symbol(watch) => vec![
+            Self::Some(items) => items
-                Property("Symbol", format!("\x1b[35m{}\x1b[0m", watch.symbol)),
+                .iter()
-                Property(
+                .map(|item| match item {
-                    "Price",
+                    InspectItem::F64(v) => format_f64(*v),
-                    format!("\x1b[96m{}\x1b[0m", format_f64(watch.price)),
+                    InspectItem::USD(v) => format!("${}", format_f64(*v)),
-                ),
+                    InspectItem::Symbol(v) => format!("\x1b[35m{v}\x1b[0m"),
-                Property("Trend", format!("{}", format_f64(watch.trend))),
+                    InspectItem::String(v) => v.clone(),
-            ]
+                })
-            .get_formatted(),
+                .collect(),
-
-            Self::Position(pos) => vec![
-                Property("Symbol", format!("\x1b[35m{}\x1b[0m", pos.symbol)),
-                Property("Profit", format!("{:+.2}", pos.profit)),
-                Property("Amount", format!("{}", pos.amount)),
-            ]
-            .get_formatted(),
-
-            Self::Signal(sig) => vec![
-                Property("Type", format!("{}", sig.kind)),
-                Property("Symbol", format!("\x1b[35m{}\x1b[0m", sig.symbol)),
-                Property("Parameter", format!("{}", sig.param)),
-                Property("Price", format!("\x1b[96m{}\x1b[0m", format_f64(sig.price))),
-            ]
-            .get_formatted(),
-
-            Self::Alert(alert) => vec![
-                Property("Level", format!("{}", alert.level)),
-                Property("Message", alert.message.clone()),
-            ]
-            .get_formatted(),
         }
     }
 }
@@ -101,7 +80,8 @@ impl Formatted for WatchListItem {
     fn get_formatted(&self) -> Vec<String> {
         vec![
             format!("\x1b[35m{}\x1b[0m", self.symbol),
-            format_f64(self.price),
+            format!("${}", format_f64(self.price)),
+            format!("${}", format_f64(self.volume_24h)),
             format!(
                 "{} {}",
                 if self.trend.is_sign_positive() {
@@ -119,7 +99,9 @@ impl Formatted for ActivePosition {
     fn get_formatted(&self) -> Vec<String> {
         vec![
             format!("\x1b[35m{}\x1b[0m", self.symbol),
-            format_f64(self.amount),
+            format_f64(self.size),
+            format_f64(self.entry_price),
+            format_f64(self.mark_price),
             format!(
                 "{} {}",
                 if self.profit.is_sign_positive() {
@@ -227,7 +209,31 @@ pub fn apply_padding(mut items: Vec<String>) -> Vec<String> {
 }
 
 pub fn format_f64(value: f64) -> String {
-    let val = value.to_string();
+    let abs = value.abs();
+
+    let (divisor, suffix) = if abs >= 1_000_000_000.0 {
+        (1_000_000_000.0, "B")
+    } else if abs >= 1_000_000.0 {
+        (1_000_000.0, "M")
+    } else if abs >= 1_000.0 {
+        (1_000.0, "K")
+    } else {
+        (1.0, "")
+    };
+
+    if divisor != 1.0 {
+        let formatted = value / divisor;
+
+        // Remove unnecessary trailing zeros
+        let s = format!("{:.2}", formatted)
+            .trim_end_matches('0')
+            .trim_end_matches('.')
+            .to_string();
+
+        return format!("{}{}", s, suffix);
+    }
+
+    let val = format!("{:.3}", value);
     let parts: Vec<&str> = val.split('.').collect();
 
     let int = parts[0].to_string();