selimaj-dev/pulse-trader
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #19 from selimaj-dev/engine-refactor

Browse Source 
Engine refactor
This commit is contained in:
selimaj-dev
2026-07-31 03:51:44 +02:00
committed by GitHub
parent 8dcf6b2a41 c9a37e4024
commit 7247319a38
18 changed files with 408 additions and 504 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.DS_Store
Vendored
BIN
View File
Binary file not shown.
Cargo.lock
Generated
+15 -14
View File
@@ -2432,9 +2432,9 @@ checksum = "e6d5a32815ae3f33302d95fdcb2ce17862f8c65363dcfd29360480ba1001fc9c"
[[package]]
name = "futures"
version = "0.3.32"
version = "0.3.33"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "8b147ee9d1f6d097cef9ce628cd2ee62288d963e16fb287bd9286455b241382d"
checksum = "a88cf1f829d945f548cf8fec32c61b1f202b6d93b45848602fc02af4b12ad218"
dependencies = [
"futures-channel",
"futures-core",
@@ -2447,9 +2447,9 @@ dependencies = [
[[package]]
name = "futures-channel"
version = "0.3.32"
version = "0.3.33"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "07bbe89c50d7a535e539b8c17bc0b49bdb77747034daa8087407d655f3f7cc1d"
checksum = "262590f4fe6afeb0bc83be1daa64e52657fe185690a958af7f3ad0e92085c5ae"
dependencies = [
"futures-core",
"futures-sink",
@@ -2457,15 +2457,15 @@ dependencies = [
[[package]]
name = "futures-core"
version = "0.3.32"
version = "0.3.33"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7e3450815272ef58cec6d564423f6e755e25379b217b0bc688e295ba24df6b1d"
checksum = "2cd50c473c80f6d7c3670a752354b8e569b1a7cbfdc0419ec88e5edad85e0dc7"
[[package]]
name = "futures-executor"
version = "0.3.32"
version = "0.3.33"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "baf29c38818342a3b26b5b923639e7b1f4a61fc5e76102d4b1981c6dc7a7579d"
checksum = "6754879cc9f2c66f88c6e5c35344bb0bdb0708b0352b1201815667c7eabc7458"
dependencies = [
"futures-core",
"futures-task",
@@ -2480,9 +2480,9 @@ checksum = "4577ecaa3c4f96589d473f679a71b596316f6641bc350038b962a5daf0085d7a"
[[package]]
name = "futures-macro"
version = "0.3.32"
version = "0.3.33"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e835b70203e41293343137df5c0664546da5745f82ec9b84d40be8336958447b"
checksum = "2d6d3cde68c518367be28956066ddfef33813991b77a55005a69dae04bf3b10b"
dependencies = [
"proc-macro2",
"quote",
@@ -2497,15 +2497,15 @@ checksum = "e34418ac499d6305c2fb5ad0ed2f6ac998c5f8ca209b4510f7f94242c647e307"
[[package]]
name = "futures-task"
version = "0.3.32"
version = "0.3.33"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "037711b3d59c33004d3856fbdc83b99d4ff37a24768fa1be9ce3538a1cde4393"
checksum = "b231ed28831efb4a61a08580c4bc233ec56bc009f4cd8f52da2c3cb97df0c109"
[[package]]
name = "futures-util"
version = "0.3.32"
version = "0.3.33"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "389ca41296e6190b48053de0321d02a77f32f8a5d2461dd38762c0593805c6d6"
checksum = "a77a90a256fce34da66415271e30f94ee91c57b04b8a2c042d9cf3220179deaa"
dependencies = [
"futures-channel",
"futures-core",
@@ -3959,6 +3959,7 @@ dependencies = [
"anyhow",
"chrono",
"crossterm",
"futures",
"hypersdk",
"postcard",
"pulse-sdk",
Cargo.toml
+1
View File
@@ -25,6 +25,7 @@ chrono = "0.4.45"
serde_json = "1"
rand = "0.8.7"
toml = "1.1.3"
futures = "0.3.33"
[workspace]
members = ["pulse-ui", "pulse-sdk"]
pulse-sdk/src/general.rs
+9 -11
View File
@@ -1,7 +1,6 @@
use hypersdk::hypercore::Side;
use rust_decimal::Decimal;
use crate::units::{Direction, Symbol, USD};
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub enum MarketTrend {
Bullish,
@@ -20,12 +19,11 @@ pub enum LogKind {
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct Signal {
pub symbol: String,
pub kind: Direction,
pub side: Side,
pub confidence: f32,
pub size: Decimal,
pub price: USD,
pub take_profit: USD,
pub stop_loss: USD,
pub entry_price: Decimal,
pub take_profit: Decimal,
pub stop_loss: Decimal,
}
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
@@ -37,10 +35,10 @@ pub struct EventLog {
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct Position {
pub symbol: Symbol,
pub size: f64,
pub entry_price: USD,
pub profit: USD,
pub symbol: String,
pub size: Decimal,
pub entry_price: Decimal,
pub pnl: Decimal,
}
impl std::fmt::Display for MarketTrend {
pulse-sdk/src/lib.rs
+15 -16
View File
@@ -1,19 +1,18 @@
pub mod general;
pub mod strategy;
pub mod terminal;
pub mod units;
pub use hypersdk;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use crate::strategy::StrategyEngineMessage;
use crate::{general::LogKind, strategy::StrategyEngineMessage};
pub mod prelude {
pub use crate::Strategy;
pub use crate::general::*;
pub use crate::strategy::*;
pub use crate::server_path;
pub use crate::strategy::*;
pub use crate::terminal::*;
pub use crate::units::*;
pub use hypersdk;
pub use postcard;
@@ -37,8 +36,8 @@ pub async fn send_raw(data: &[u8]) -> tokio::io::Result<()> {
Ok(())
}
macro_rules! engine_methods {
($t:ty) => {
#[allow(async_fn_in_trait)]
pub trait Strategy {
async fn start(&self) -> tokio::io::Result<()> {
let mut stdin = tokio::io::stdin();
@@ -60,18 +59,18 @@ macro_rules! engine_methods {
}
}
async fn send(&self, msg: &$t) -> tokio::io::Result<()> {
$crate::send_raw(&$crate::map_postcard_err(
$crate::prelude::postcard::to_allocvec(msg),
)?)
.await
}
};
async fn send(&self, msg: &prelude::StrategyMessage) -> tokio::io::Result<()> {
send_raw(&map_postcard_err(prelude::postcard::to_allocvec(msg))?).await
}
#[allow(async_fn_in_trait)]
pub trait Strategy {
engine_methods!(prelude::StrategyMessage);
async fn log(&self, kind: LogKind, name: &str, message: &str) -> tokio::io::Result<()> {
self.send(&prelude::StrategyMessage::Log(prelude::EventLog {
kind,
name: name.to_owned(),
message: message.to_owned(),
}))
.await
}
async fn on_raw(&self, data: &[u8]) -> tokio::io::Result<()> {
match map_postcard_err(postcard::from_bytes(&data))? {
pulse-sdk/src/terminal.rs
+6 -44
View File
@@ -1,7 +1,6 @@
use crate::{
general::{EventLog, MarketTrend, Position, Signal},
general::{EventLog, Position, Signal},
strategy::StrategyManifest,
units::{Symbol, USD, Volatility},
};
use hypersdk::{Decimal, hypercore::CandleInterval};
@@ -16,7 +15,7 @@ pub enum TerminalServerMessage {
PositionsUpdated(Vec<Position>),
// Strategy
StrategyUpdated(Strategy),
StrategyUpdated(StrategyStatus),
// Signals
SignalsUpdated(Vec<SignalStatus>),
@@ -39,37 +38,10 @@ pub enum TerminalClientMessage {
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct MarketItem {
pub symbol: Symbol,
pub price: USD,
pub symbol: String,
pub price: Decimal,
pub trend: Decimal,
pub volume_24h: USD,
}
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct MarketOverview {
pub trend: MarketTrend,
pub volatility: Volatility,
pub pressure: f64,
pub alerts: Vec<Alert>,
}
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub enum AlertLevel {
High,
Medium,
Low,
}
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct Alert {
pub level: AlertLevel,
pub message: String,
}
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct Balance {
pub asset: String,
pub amount: f64,
pub value: f64,
pub volume_24h: Decimal,
}
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
@@ -108,7 +80,7 @@ pub enum ItemState {
}
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct Strategy {
pub struct StrategyStatus {
pub strategy: StrategyManifest,
pub mode: Mode,
@@ -116,16 +88,6 @@ pub struct Strategy {
pub cooldown: CandleInterval,
}
impl std::fmt::Display for AlertLevel {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::High => write!(f, "H"),
Self::Medium => write!(f, "M"),
Self::Low => write!(f, "L"),
}
}
}
impl std::fmt::Display for Mode {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
pulse-sdk/src/units.rs
-110
View File
@@ -1,110 +0,0 @@
use rust_decimal::Decimal;
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct Symbol(pub String);
#[derive(Debug, Clone, Copy, serde::Serialize, serde::Deserialize)]
pub struct USD(pub Decimal);
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub enum Direction {
Buy,
Sell,
}
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub enum Volatility {
Low,
Medium,
High,
}
impl std::fmt::Display for Symbol {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "\x1b[35m{}\x1b[0m", self.0)
}
}
impl std::fmt::Display for USD {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
if self.0.is_sign_positive() {
write!(f, "\x1b[32m${}\x1b[0m", format_f64(self.0.as_f64()))
} else {
write!(f, "\x1b[31m${}\x1b[0m", format_f64(self.0.as_f64()))
}
}
}
pub fn format_f64(value: f64) -> String {
let abs = value.abs();
let (divisor, suffix) = if abs >= 1_000_000_000.0 {
(1_000_000_000.0, "B")
} else if abs >= 1_000_000.0 {
(1_000_000.0, "M")
} else if abs >= 1_000.0 {
(1_000.0, "K")
} else {
(1.0, "")
};
if divisor != 1.0 {
let formatted = value / divisor;
// Remove unnecessary trailing zeros
let s = format!("{:.2}", formatted)
.trim_end_matches('0')
.trim_end_matches('.')
.to_string();
return format!("{}{}", s, suffix);
}
let val = format!("{:.3}", value);
let parts: Vec<&str> = val.split('.').collect();
let int = parts[0].to_string();
let negative = int.starts_with('-');
let start = if negative { 1 } else { 0 };
let mut result = String::new();
for (i, c) in int[start..].chars().rev().enumerate() {
if i > 0 && i % 3 == 0 {
result.push(',');
}
result.push(c);
}
let mut formatted: String = result.chars().rev().collect();
if negative {
formatted.insert(0, '-');
}
if parts.len() > 1 {
formatted.push('.');
formatted.push_str(parts[1]);
}
formatted
}
impl std::fmt::Display for Direction {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::Buy => write!(f, "BUY"),
Self::Sell => write!(f, "SELL"),
}
}
}
impl std::fmt::Display for Volatility {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::Low => write!(f, "Low"),
Self::Medium => write!(f, "Medium"),
Self::High => write!(f, "High"),
}
}
}
src/engine/engine/command.rs
+27 -6
View File
@@ -1,10 +1,18 @@
use crate::{engine::Engine, store::config::Config};
use pulse_sdk::terminal::{ItemState, Mode, Strategy};
use pulse_sdk::prelude::*;
use toml::Value;
impl Engine {
pub async fn execute_command(&self, command: &str, args: Vec<&str>) -> tokio::io::Result<()> {
pub async fn invalid_command_usage(&self, name: &str) -> tokio::io::Result<()> {
self.terminal_server.error(name, "Invalid usage").await
}
pub async fn execute_command(
&self,
command: &str,
mut args: Vec<&str>,
) -> tokio::io::Result<()> {
match command {
"config" | "cfg" => {
if args.len() == 0 {
@@ -43,7 +51,7 @@ impl Engine {
.await?;
}
"strategy" | "strat" | "str" | "sg" => {
"strategy" | "strat" | "sg" => {
set_cfg!(id, {
let id: String = id;
@@ -60,7 +68,7 @@ impl Engine {
.await;
}
self.strategy.reload_strategy(id.as_str()).await?;
self.strategy_engine.reload(id.as_str()).await?;
self.config.lock().await.strategy = id;
});
@@ -74,8 +82,8 @@ impl Engine {
self.config.lock().await.cooldown = cooldown;
self.terminal_server.broadcast(
pulse_sdk::terminal::TerminalServerMessage::StrategyUpdated(Strategy {
strategy: self.strategy.strategy.manifest.lock().await.clone(),
pulse_sdk::terminal::TerminalServerMessage::StrategyUpdated(StrategyStatus {
strategy: self.strategy_engine.strategy.lock().await.manifest.clone(),
mode: Mode::Auto,
state: ItemState::Running,
@@ -166,6 +174,19 @@ impl Engine {
}
}
"strategy" | "strat" | "sg" => {
let Some(strategy_command) = args.drain(0..=0).next() else {
return self.invalid_command_usage("strategy").await;
};
self.strategy_engine
.send(&StrategyEngineMessage::Command {
command: strategy_command.to_owned(),
args: args.into_iter().map(Into::into).collect(),
})
.await?;
}
_ => {
self.terminal_server
.error(
src/engine/engine/execution.rs
+12 -12
View File
@@ -1,4 +1,4 @@
use hypersdk::hypercore::{self, BatchOrder, OrderRequest, OrderTypePlacement, TimeInForce};
use hypersdk::hypercore::{self, BatchOrder, OrderRequest, OrderTypePlacement, Side, TimeInForce};
use pulse_sdk::prelude::*;
use crate::engine::Engine;
@@ -34,9 +34,9 @@ impl Engine {
orders: vec![
OrderRequest {
asset: asset_id,
is_buy: matches!(signal.kind, Direction::Buy),
limit_px: signal.price.0,
sz: signal.size,
is_buy: matches!(signal.side, Side::Bid),
limit_px: signal.entry_price,
sz: 0.into(),
reduce_only: false,
order_type: OrderTypePlacement::Limit {
tif: TimeInForce::Gtc,
@@ -45,26 +45,26 @@ impl Engine {
},
OrderRequest {
asset: asset_id,
is_buy: matches!(signal.kind, Direction::Buy),
limit_px: signal.price.0,
sz: signal.size,
is_buy: matches!(signal.side, Side::Bid),
limit_px: signal.entry_price,
sz: 0.into(),
reduce_only: true,
order_type: OrderTypePlacement::Trigger {
is_market: true,
trigger_px: signal.take_profit.0,
trigger_px: signal.take_profit,
tpsl: hypercore::TpSl::Tp,
},
cloid: Default::default(),
},
OrderRequest {
asset: asset_id,
is_buy: matches!(signal.kind, Direction::Buy),
limit_px: signal.price.0,
sz: signal.size,
is_buy: matches!(signal.side, Side::Bid),
limit_px: signal.entry_price,
sz: 0.into(),
reduce_only: true,
order_type: OrderTypePlacement::Trigger {
is_market: true,
trigger_px: signal.stop_loss.0,
trigger_px: signal.stop_loss,
tpsl: hypercore::TpSl::Sl,
},
cloid: Default::default(),
src/engine/engine/mod.rs
+20 -86
View File
@@ -1,15 +1,16 @@
pub mod command;
pub mod execution;
pub mod strategy;
pub mod terminal;
use crate::{
engine::strategy::StrategyEngine,
engine::{strategy::StrategyEngine, terminal::TerminalServer},
store::{accounts::AccountList, config::Config},
terminal::TerminalServer,
};
use hypersdk::hypercore::ws::ConnectionStream;
use pulse_sdk::prelude::*;
use std::{collections::HashMap, sync::Arc};
use tokio::{sync::Mutex, task::JoinHandle};
use tokio::sync::Mutex;
#[derive(Debug, Clone)]
pub struct WatchList {
@@ -19,8 +20,12 @@ pub struct WatchList {
#[derive(Clone)]
pub struct Engine {
// engine
pub terminal_server: Arc<TerminalServer>,
pub strategy: Arc<StrategyEngine>,
pub strategy_engine: Arc<StrategyEngine>,
pub ws_stream: Arc<Mutex<ConnectionStream>>,
// data
pub config: Arc<Mutex<Config>>,
pub accounts: Arc<Mutex<AccountList>>,
pub watch_list: Arc<Mutex<WatchList>>,
@@ -31,16 +36,19 @@ impl Engine {
pub async fn new() -> tokio::io::Result<Arc<Self>> {
let config = Config::new().await?;
let strategy = StrategyEngine::new(&config.strategy).await?;
let (ws_handle, ws_stream) = hypersdk::hypercore::mainnet_ws().split();
let strategy = StrategyEngine::new(&config.strategy, ws_handle).await?;
let accounts = Arc::new(Mutex::new(AccountList::new().await?));
let config = Arc::new(Mutex::new(config));
Ok(Arc::new_cyclic(|engine| Self {
terminal_server: TerminalServer::new(engine.clone()),
strategy: strategy.initialize(engine.clone()),
config,
accounts,
ws_stream: Arc::new(Mutex::new(ws_stream)),
terminal_server: TerminalServer::new(engine.clone()),
strategy_engine: strategy.initialize(engine.clone()),
watch_list: Arc::new(Mutex::new(WatchList {
name_to_index: HashMap::new(),
items: Vec::new(),
@@ -49,89 +57,15 @@ impl Engine {
}))
}
pub async fn spawn_broadcaster(&self) -> JoinHandle<tokio::io::Result<()>> {
let s = self.clone();
tokio::spawn(async move { s.run_broadcaster().await })
}
pub async fn run_broadcaster(&self) -> tokio::io::Result<()> {
let mut refresh = tokio::time::interval(tokio::time::Duration::from_secs(5));
let client = hypersdk::hypercore::mainnet();
/// Starts the main strategy server
/// When a strategy is reloaded it restarts automatically
pub async fn run(&self) -> anyhow::Result<()> {
loop {
refresh.tick().await;
self.strategy_engine.run().await?;
let watch_list = &self.config.lock().await.watchlist;
match crate::fetch::fetch_watch_list(&client, watch_list).await {
Ok(watch_list) => {
*self.watch_list.lock().await = watch_list.clone();
if let Err(error) = self
.terminal_server
.broadcast(TerminalServerMessage::WatchListUpdated(watch_list.items))
.await
{
self.terminal_server
.error(
"Broadcaster",
&format!("Failed to broadcast HyperLiquid watch list: {error}"),
)
.await?
}
}
Err(error) => {
self.terminal_server
.error(
"Broadcaster",
&format!("Failed to refresh HyperLiquid watch list: {error}"),
)
.await?
}
}
let accounts = self.accounts.lock().await;
if let Some(acc) = accounts.get_active() {
match client.clearinghouse_state(acc.address, None).await {
Ok(state) => {
self.terminal_server
.broadcast(TerminalServerMessage::PositionsUpdated(
state
.asset_positions
.into_iter()
.map(|position| Position {
symbol: Symbol(position.position.coin),
size: position.position.szi.as_f64(),
entry_price: USD(position
.position
.entry_px
.unwrap_or_default()),
profit: USD(position.position.unrealized_pnl),
})
.collect(),
))
.await?;
}
Err(e) => {
self.terminal_server
.error("orders", &format!("Unable to get open orders: {e}"))
.info("engine::main", "Strategy stopped, restarting")
.await?;
}
}
} else {
self.terminal_server.error(
"orders",
"Unable to get active account, make sure you have configured accounts properly",
).await?;
}
}
}
pub async fn invalid_command_usage(&self, name: &str) -> tokio::io::Result<()> {
self.terminal_server.error(name, "Invalid usage").await
}
}
src/engine/engine/strategy.rs
+59 -69
View File
@@ -1,45 +1,48 @@
use hypersdk::hypercore::{self, CandleInterval, Subscription, WebSocket};
use anyhow::Context;
use hypersdk::hypercore::{self, CandleInterval, Subscription, ws::ConnectionHandle};
use pulse_sdk::prelude::*;
use std::{
collections::HashSet,
path::PathBuf,
process::Stdio,
sync::{Arc, Weak},
time::{SystemTime, UNIX_EPOCH},
};
use tokio::{
fs,
process::{Child, Command},
sync::Mutex,
};
use tokio::{fs, sync::Mutex};
use crate::{
engine::Engine,
store::{pulse_strategy, strategy::StrategyChild},
store::{
pulse_strategy,
strategy::{StrategyChild, get_manifest},
},
};
pub struct StrategyEngine {
pub strategy: Arc<StrategyChild>,
pub engine: Weak<Engine>,
pub ws: WebSocket,
pub strategy: Mutex<StrategyChild>,
pub ws_handle: ConnectionHandle,
pub subscriptions: Mutex<HashSet<Subscription>>,
}
impl StrategyEngine {
pub async fn new(strategy_id: &str) -> tokio::io::Result<Self> {
pub async fn new(strategy_id: &str, ws_handle: ConnectionHandle) -> tokio::io::Result<Self> {
let strategy = pulse_strategy(strategy_id)?;
let (strategy, strategy_manifest) = get_manifest(
&strategy,
&strategy.join("strategy.bash"),
&fs::read(strategy.join("strategy.toml")).await?,
&match fs::read(strategy.join("strategy.toml")).await {
Err(e) if e.kind() == tokio::io::ErrorKind::NotFound => Err(tokio::io::Error::new(
tokio::io::ErrorKind::NotFound,
format!("Unable to find strategy.toml for {strategy:?}: {}", e),
)),
v => v,
}?,
)?;
Ok(Self {
strategy: Arc::new(StrategyChild::new(strategy, strategy_manifest)),
strategy: Mutex::new(StrategyChild::new(strategy, strategy_manifest)),
engine: Weak::new(),
ws: hypercore::mainnet_ws(),
ws_handle,
subscriptions: Mutex::new(HashSet::new()),
})
}
@@ -50,23 +53,52 @@ impl StrategyEngine {
Arc::new(self)
}
pub async fn run_strategy(&self) -> anyhow::Result<()> {
pub async fn send(self: &Arc<Self>, msg: &StrategyEngineMessage) -> tokio::io::Result<()> {
self.strategy.lock().await.send(msg).await
}
pub async fn reload(self: &Arc<Self>, id: &str) -> tokio::io::Result<()> {
let strategy = pulse_strategy(id)?;
let (child, manifest) = get_manifest(
&strategy,
&strategy.join("strategy.bash"),
&fs::read(strategy.join("strategy.toml")).await?,
)?;
let mut strategy = self.strategy.lock().await;
strategy.child.kill().await?;
strategy.child = child;
strategy.manifest = manifest;
Ok(())
}
pub async fn run(self: &Arc<Self>) -> anyhow::Result<()> {
let engine = self
.engine
.upgrade()
.expect("Failed to upgrade engine (StrategyEngine)");
self.strategy
.send(&StrategyEngineMessage::Initialize)
.await?;
let mut stdout = {
let mut child = self.strategy.lock().await;
child
.child
.stdout
.take()
.context("Failed to obtain strategy stdout")?
};
self.send(&StrategyEngineMessage::Initialize).await?;
loop {
match self.strategy.recv().await? {
match StrategyChild::read(&mut stdout).await? {
None => {}
Some(StrategyMessage::GetWatchList) => {
self.strategy
.send(&StrategyEngineMessage::WatchList(
self.send(&StrategyEngineMessage::WatchList(
engine.watch_list.lock().await.clone().items,
))
.await?;
@@ -99,18 +131,18 @@ impl StrategyEngine {
}
Some(StrategyMessage::Subscribe(subscription)) => {
self.ws.subscribe(subscription.clone());
self.ws_handle.subscribe(subscription.clone());
self.subscriptions.lock().await.insert(subscription);
}
Some(StrategyMessage::Unsubscribe(subscription)) => {
self.subscriptions.lock().await.remove(&subscription);
self.ws.unsubscribe(subscription);
self.ws_handle.unsubscribe(subscription);
}
Some(StrategyMessage::UnsubscribeAll) => {
for sub in self.subscriptions.lock().await.drain() {
self.ws.unsubscribe(sub);
self.ws_handle.unsubscribe(sub);
}
}
@@ -147,8 +179,7 @@ impl StrategyEngine {
let start_time = now.saturating_sub(interval_ms * count as u64);
self.strategy
.send(&StrategyEngineMessage::Candlestick {
self.send(&StrategyEngineMessage::Candlestick {
candles: client
.candle_snapshot(&symbol, interval, start_time, now)
.await?,
@@ -160,45 +191,4 @@ impl StrategyEngine {
}
}
}
pub async fn spawn(self: &Arc<Self>) {
let engine = self.clone();
tokio::spawn(async move { engine.run_strategy().await });
}
pub async fn reload_strategy(self: &Arc<Self>, id: &str) -> tokio::io::Result<()> {
let strategy = pulse_strategy(id)?;
let (child, manifest) = get_manifest(
&strategy,
&strategy.join("strategy.bash"),
&fs::read(strategy.join("strategy.toml")).await?,
)?;
self.strategy.reload(child, manifest).await?;
let engine = self.clone();
tokio::spawn(async move { engine.run_strategy().await });
Ok(())
}
}
fn get_manifest<'de, M: serde::Deserialize<'de>>(
strategy_dir: &PathBuf,
strategy_path: &PathBuf,
manifest: &'de [u8],
) -> tokio::io::Result<(Child, M)> {
Ok((
Command::new("bash")
.arg(strategy_path)
.current_dir(strategy_dir)
.stdin(Stdio::piped())
.stdout(Stdio::piped())
.stderr(Stdio::inherit())
.spawn()?,
toml::from_slice(manifest)
.map_err(|v| tokio::io::Error::new(std::io::ErrorKind::InvalidInput, v.to_string()))?,
))
}
src/engine/terminal.rs → src/engine/engine/terminal.rs
+91 -3
View File
@@ -11,6 +11,7 @@ use tokio::{
unix::{OwnedReadHalf, OwnedWriteHalf},
},
sync::Mutex,
task::JoinHandle,
};
#[derive(Debug)]
@@ -29,7 +30,13 @@ impl TerminalServer {
})
}
pub async fn run(self: &Arc<Self>) -> tokio::io::Result<()> {
pub async fn spawn_server(self: &Arc<Self>) -> JoinHandle<tokio::io::Result<()>> {
let s = self.clone();
tokio::spawn(s.run_server())
}
pub async fn run_server(self: Arc<Self>) -> tokio::io::Result<()> {
let path = pulse_sdk::server_path();
if path.exists() {
@@ -64,8 +71,14 @@ impl TerminalServer {
self.send_to(
id,
pulse_sdk::terminal::TerminalServerMessage::StrategyUpdated(Strategy {
strategy: engine.strategy.strategy.manifest.lock().await.clone(),
pulse_sdk::terminal::TerminalServerMessage::StrategyUpdated(StrategyStatus {
strategy: engine
.strategy_engine
.strategy
.lock()
.await
.manifest
.clone(),
mode: Mode::Auto,
state: ItemState::Running,
cooldown: engine.config.lock().await.cooldown,
@@ -145,6 +158,81 @@ impl TerminalServer {
Ok(())
}
pub async fn spawn_broadcaster(self: &Arc<Self>) -> JoinHandle<tokio::io::Result<()>> {
let s = self.clone();
tokio::spawn(s.run_broadcaster())
}
pub async fn run_broadcaster(self: Arc<Self>) -> tokio::io::Result<()> {
let engine = self.get_engine();
let mut refresh = tokio::time::interval(tokio::time::Duration::from_secs(5));
let client = hypersdk::hypercore::mainnet();
loop {
refresh.tick().await;
match crate::fetch::fetch_watch_list(&client, &engine.config.lock().await.watchlist)
.await
{
Ok(watch_list) => {
*engine.watch_list.lock().await = watch_list.clone();
if let Err(error) = self
.broadcast(TerminalServerMessage::WatchListUpdated(watch_list.items))
.await
{
self.error(
"Broadcaster",
&format!("Failed to broadcast HyperLiquid watch list: {error}"),
)
.await?
}
}
Err(error) => {
self.error(
"Broadcaster",
&format!("Failed to refresh HyperLiquid watch list: {error}"),
)
.await?
}
}
if let Some(acc) = engine.accounts.lock().await.get_active() {
match client.clearinghouse_state(acc.address, None).await {
Ok(state) => {
self.broadcast(TerminalServerMessage::PositionsUpdated(
state
.asset_positions
.into_iter()
.map(|position| Position {
symbol: position.position.coin,
size: position.position.szi,
entry_price: position.position.entry_px.unwrap_or_default(),
pnl: position.position.unrealized_pnl,
})
.collect(),
))
.await?;
}
Err(e) => {
self.error("orders", &format!("Unable to get open orders: {e}"))
.await?;
}
}
} else {
self.error(
"orders",
"Unable to get active account, make sure you have configured accounts properly",
)
.await?;
}
}
}
pub async fn send_to(
self: &Arc<Self>,
id: &usize,
src/engine/fetch.rs
+4 -4
View File
@@ -1,5 +1,5 @@
use rust_decimal::Decimal;
use pulse_sdk::prelude::*;
use rust_decimal::Decimal;
use serde_json::Value;
use std::collections::HashMap;
@@ -80,9 +80,9 @@ pub async fn fetch_watch_list(
}
items.push(MarketItem {
symbol: Symbol(symbol.clone()),
price: USD(price),
volume_24h: USD(volume_24h),
symbol: symbol.clone(),
price: price,
volume_24h: volume_24h,
trend: ((price / previous_day_price) - <Decimal as From<i32>>::from(1))
* <Decimal as From<i32>>::from(100),
});
src/engine/main.rs
+4 -5
View File
@@ -1,18 +1,17 @@
pub mod engine;
pub mod fetch;
pub mod store;
pub mod terminal;
#[tokio::main]
async fn main() -> anyhow::Result<()> {
let engine = engine::Engine::new().await?;
let broadcaster = engine.spawn_broadcaster().await;
let server = engine.terminal_server.spawn_server().await;
let broadcaster = engine.terminal_server.spawn_broadcaster().await;
engine.strategy.spawn().await;
engine.terminal_server.run().await?;
engine.run().await?;
server.await??;
broadcaster.await??;
Ok(())
src/engine/store/strategy.rs
+28 -33
View File
@@ -1,31 +1,25 @@
use std::{path::PathBuf, process::Stdio};
use pulse_sdk::{
map_postcard_err,
strategy::{StrategyEngineMessage, StrategyManifest, StrategyMessage},
};
use tokio::{
io::{AsyncReadExt, AsyncWriteExt},
process::{Child, ChildStdout},
sync::Mutex,
process::{Child, ChildStdout, Command},
};
#[derive(Debug)]
pub struct StrategyChild {
pub manifest: Mutex<StrategyManifest>,
pub stdout: Mutex<ChildStdout>,
pub process: Mutex<Child>,
pub manifest: StrategyManifest,
pub child: Child,
}
impl StrategyChild {
pub fn new(mut child: Child, manifest: StrategyManifest) -> Self {
Self {
stdout: Mutex::new(child.stdout.take().expect("Failed to obtain child stdout")),
process: Mutex::new(child),
manifest: Mutex::new(manifest),
}
pub fn new(child: Child, manifest: StrategyManifest) -> Self {
Self { child, manifest }
}
pub async fn recv(&self) -> tokio::io::Result<Option<StrategyMessage>> {
let mut stdout = self.stdout.lock().await;
pub async fn read(stdout: &mut ChildStdout) -> tokio::io::Result<Option<StrategyMessage>> {
let mut len_buf = [0u8; size_of::<usize>()];
let size = stdout.read_exact(&mut len_buf).await?;
@@ -42,14 +36,13 @@ impl StrategyChild {
Ok(Some(map_postcard_err(postcard::from_bytes(&buffer))?))
}
pub async fn send(&self, msg: &StrategyEngineMessage) -> tokio::io::Result<()> {
pub async fn send(&mut self, msg: &StrategyEngineMessage) -> tokio::io::Result<()> {
self.send_raw(&map_postcard_err(postcard::to_allocvec(msg))?)
.await
}
pub async fn send_raw(&self, msg: &[u8]) -> tokio::io::Result<()> {
let mut process = self.process.lock().await;
let stdin = process.stdin.as_mut().unwrap();
pub async fn send_raw(&mut self, msg: &[u8]) -> tokio::io::Result<()> {
let stdin = self.child.stdin.as_mut().unwrap();
stdin.write_all(&msg.len().to_le_bytes()).await?;
stdin.write_all(msg).await?;
@@ -57,20 +50,22 @@ impl StrategyChild {
Ok(())
}
pub async fn reload(
&self,
mut child: Child,
manifest: StrategyManifest,
) -> tokio::io::Result<()> {
let mut process = self.process.lock().await;
process.kill().await?;
*self.stdout.lock().await = child.stdout.take().expect("Failed to obtain child stdout");
*self.manifest.lock().await = manifest;
*process = child;
Ok(())
}
pub fn get_manifest<'de, M: serde::Deserialize<'de>>(
strategy_dir: &PathBuf,
strategy_path: &PathBuf,
manifest: &'de [u8],
) -> tokio::io::Result<(Child, M)> {
Ok((
Command::new("bash")
.arg(strategy_path)
.current_dir(strategy_dir)
.stdin(Stdio::piped())
.stdout(Stdio::piped())
.stderr(Stdio::inherit())
.spawn()?,
toml::from_slice(manifest)
.map_err(|v| tokio::io::Error::new(std::io::ErrorKind::InvalidInput, v.to_string()))?,
))
}
src/terminal/formatting.rs
+82 -56
View File
@@ -1,3 +1,4 @@
use hypersdk::hypercore::Side;
use pulse_sdk::prelude::*;
pub trait Formatted {
@@ -22,23 +23,6 @@ impl Formatted for InspectTarget {
}
}
impl Formatted for Alert {
fn get_formatted(&self) -> Vec<String> {
vec![
format!(
"{}{}\x1b[0m",
match &self.level {
AlertLevel::Low => "\x1b[34m",
AlertLevel::Medium => "\x1b[33m",
AlertLevel::High => "\x1b[31m",
},
self.level
),
self.message.clone(),
]
}
}
impl Formatted for EventLog {
fn get_formatted(&self) -> Vec<String> {
vec![
@@ -64,13 +48,13 @@ impl Formatted for SignalStatus {
Ok(signal) => {
vec![
format!("\x1b[33mOK\x1b[0m"),
if matches!(signal.kind, Direction::Buy) {
if matches!(signal.side, Side::Ask) {
format!("\x1b[32mBUY\x1b[0m")
} else {
format!("\x1b[31mSELL\x1b[0m")
},
format!("\x1b[35m{}\x1b[0m", signal.symbol),
signal.price.to_string(),
format_symbol(&signal.symbol),
format_usd(signal.entry_price.as_f64()),
format!("\x1b[33mAPR {}\x1b[0m", signal.confidence),
]
}
@@ -78,13 +62,13 @@ impl Formatted for SignalStatus {
Err(signal) => {
vec![
format!("\x1b[31mERR\x1b[0m"),
if matches!(signal.kind, Direction::Buy) {
if matches!(signal.side, Side::Ask) {
format!("\x1b[32mBUY\x1b[0m")
} else {
format!("\x1b[31mSELL\x1b[0m")
},
format!("\x1b[35m{}\x1b[0m", signal.symbol),
signal.price.to_string(),
format_symbol(&signal.symbol),
format_usd(signal.entry_price.as_f64()),
format!("\x1b[33mAPR {}\x1b[0m", signal.confidence),
]
}
@@ -95,9 +79,9 @@ impl Formatted for SignalStatus {
impl Formatted for MarketItem {
fn get_formatted(&self) -> Vec<String> {
vec![
self.symbol.to_string(),
self.price.to_string(),
self.volume_24h.to_string(),
format_symbol(&self.symbol),
format_usd(self.price.as_f64()),
format_usd(self.volume_24h.as_f64()),
format!(
"{} {}%",
if self.trend.is_sign_positive() {
@@ -114,17 +98,17 @@ impl Formatted for MarketItem {
impl Formatted for Position {
fn get_formatted(&self) -> Vec<String> {
vec![
self.symbol.to_string(),
format_f64(self.size),
format_symbol(&self.symbol),
format_f64(self.size.as_f64()),
self.entry_price.to_string(),
format!(
"{}{}\x1b[0m",
if self.profit.0.is_sign_positive() {
if self.pnl.is_sign_positive() {
"\x1b[32m"
} else {
"\x1b[31m"
},
self.profit,
self.pnl,
),
]
}
@@ -134,31 +118,6 @@ struct Pair<'a>(&'a str, &'a str);
struct Triple<'a>(&'a str, &'a str, &'a str);
impl Formatted for MarketOverview {
fn get_formatted(&self) -> Vec<String> {
let mut o = vec![
Pair("TREND", &format!("{}", self.trend)),
Pair("VOLATILITY", &format!("{}", self.volatility)),
Pair(
"PRESSURE",
&if self.pressure.is_sign_positive() {
format!("\x1b[32m{:.3}\x1b[0m", self.pressure)
} else {
format!("\x1b[31m{:.3}\x1b[0m", self.pressure)
},
),
]
.get_formatted();
o.push(format!(
"\n{}",
apply_padding(self.alerts.get_formatted()).join("\n")
));
o
}
}
impl Formatted for Status {
fn get_formatted(&self) -> Vec<String> {
vec![
@@ -247,7 +206,7 @@ impl<T: Formatted> Formatted for Vec<T> {
}
}
impl Formatted for Strategy {
impl Formatted for StrategyStatus {
fn get_formatted(&self) -> Vec<String> {
vec![
Triple(
@@ -272,3 +231,70 @@ pub fn apply_padding(mut items: Vec<String>) -> Vec<String> {
items
}
pub fn format_symbol(value: &str) -> String {
format!("\x1b[35m{value}\x1b[0m")
}
pub fn format_usd(value: f64) -> String {
if value.is_sign_positive() {
format!("\x1b[32m${}\x1b[0m", format_f64(value))
} else {
format!("\x1b[31m${}\x1b[0m", format_f64(value))
}
}
pub fn format_f64(value: f64) -> String {
let abs = value.abs();
let (divisor, suffix) = if abs >= 1_000_000_000.0 {
(1_000_000_000.0, "B")
} else if abs >= 1_000_000.0 {
(1_000_000.0, "M")
} else if abs >= 1_000.0 {
(1_000.0, "K")
} else {
(1.0, "")
};
if divisor != 1.0 {
let formatted = value / divisor;
// Remove unnecessary trailing zeros
let s = format!("{:.2}", formatted)
.trim_end_matches('0')
.trim_end_matches('.')
.to_string();
return format!("{}{}", s, suffix);
}
let val = format!("{:.3}", value);
let parts: Vec<&str> = val.split('.').collect();
let int = parts[0].to_string();
let negative = int.starts_with('-');
let start = if negative { 1 } else { 0 };
let mut result = String::new();
for (i, c) in int[start..].chars().rev().enumerate() {
if i > 0 && i % 3 == 0 {
result.push(',');
}
result.push(c);
}
let mut formatted: String = result.chars().rev().collect();
if negative {
formatted.insert(0, '-');
}
if parts.len() > 1 {
formatted.push('.');
formatted.push_str(parts[1]);
}
formatted
}
src/terminal/main.rs
+1 -1
View File
@@ -29,7 +29,7 @@ pub struct PulseTradeApp {
command: State<InputState>,
scroll: State<ScrollState<7>>,
strategy: State<Option<Strategy>>,
strategy: State<Option<StrategyStatus>>,
status: State<Option<Status>>,
watch_list: State<Vec<MarketItem>>,
src/terminal/terminal.rs
+1 -1
View File
@@ -82,7 +82,7 @@ impl TerminalClient {
active_positions: State<Vec<Position>>,
logs: State<Vec<EventLog>>,
signals: State<Vec<SignalStatus>>,
market_overview: State<Option<Strategy>>,
market_overview: State<Option<StrategyStatus>>,
status: State<Option<Status>>,
inspect: State<InspectTarget>,
) -> tokio::io::Result<()> {