Merge pull request #21 from selimaj-dev/working-strategy-and-engine

Working strategy and engine
This commit is contained in:
2026-08-02 07:08:05 +02:00
committed by GitHub
15 changed files with 954 additions and 296 deletions
+96 -2
View File
@@ -1,4 +1,7 @@
use hypersdk::hypercore::Side; use hypersdk::{
dec,
hypercore::{Cloid, Side},
};
use rust_decimal::Decimal; use rust_decimal::Decimal;
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
@@ -30,14 +33,31 @@ pub enum Mode {
Manual, Manual,
} }
#[derive(Debug, Clone, Copy)]
pub enum Allocation {
Fixed(Decimal),
Percent(Decimal),
}
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct Signal { pub struct Signal {
pub symbol: String, pub symbol: String,
pub side: Side, pub side: Side,
pub confidence: f32, /// 0-100
pub confidence: u8,
pub entry_price: Decimal, pub entry_price: Decimal,
pub take_profit: 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 stop_loss: Decimal,
pub entry: Cloid,
pub take_profit: Cloid,
pub stop_loss_id: Cloid,
} }
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
@@ -57,8 +77,10 @@ pub struct Position {
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct EngineStatus { pub struct EngineStatus {
pub perps_equity: Decimal,
pub strategy_mode: Mode, pub strategy_mode: Mode,
pub strategy_state: ItemState, pub strategy_state: ItemState,
pub positions: Vec<Position>,
} }
impl std::fmt::Display for MarketTrend { 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<D>(deserializer: D) -> Result<Self, D::Error>
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<E>(self, value: f64) -> Result<Self::Value, E>
where
E: serde::de::Error,
{
Ok(Allocation::Fixed(Decimal::from_f64_retain(value).unwrap()))
}
fn visit_str<E>(self, value: &str) -> Result<Self::Value, E>
where
E: serde::de::Error,
{
if let Some(percent) = value.strip_suffix('%') {
let value = percent.parse::<Decimal>().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<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
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}%"),
}
}
}
+18 -1
View File
@@ -3,6 +3,8 @@ pub mod strategy;
pub mod terminal; pub mod terminal;
pub use hypersdk; pub use hypersdk;
use hypersdk::Decimal;
use rust_decimal::RoundingStrategy;
use tokio::io::{AsyncReadExt, AsyncWriteExt}; use tokio::io::{AsyncReadExt, AsyncWriteExt};
use crate::{general::LogKind, strategy::StrategyEngineMessage}; use crate::{general::LogKind, strategy::StrategyEngineMessage};
@@ -10,9 +12,9 @@ use crate::{general::LogKind, strategy::StrategyEngineMessage};
pub mod prelude { pub mod prelude {
pub use crate::Strategy; pub use crate::Strategy;
pub use crate::general::*; pub use crate::general::*;
pub use crate::server_path;
pub use crate::strategy::*; pub use crate::strategy::*;
pub use crate::terminal::*; pub use crate::terminal::*;
pub use crate::{round_price, server_path};
pub use hypersdk; pub use hypersdk;
pub use postcard; pub use postcard;
@@ -26,6 +28,21 @@ 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)) 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<()> { pub async fn send_raw(data: &[u8]) -> tokio::io::Result<()> {
let mut stdout = tokio::io::stdout(); let mut stdout = tokio::io::stdout();
+7 -5
View File
@@ -1,19 +1,16 @@
use crate::{ use crate::{
general::{EngineStatus, EventLog, Position, Signal}, general::{EngineOrder, EngineStatus, EventLog, Signal},
strategy::StrategyManifest, strategy::StrategyManifest,
}; };
use hypersdk::Decimal; use hypersdk::Decimal;
pub type SignalStatus = Result<Signal, Signal>; pub type SignalStatus = Result<(Signal, Option<EngineOrder>), (Signal, Option<EngineOrder>)>;
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub enum TerminalServerMessage { pub enum TerminalServerMessage {
// WatchList // WatchList
WatchListUpdated(Vec<MarketItem>), WatchListUpdated(Vec<MarketItem>),
// Positions
PositionsUpdated(Vec<Position>),
// Configuration // Configuration
ConfigUpdated(EngineConfig), ConfigUpdated(EngineConfig),
@@ -63,5 +60,10 @@ pub enum InspectItem {
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct EngineConfig { pub struct EngineConfig {
pub preset: String,
pub strategy: StrategyManifest, pub strategy: StrategyManifest,
pub description: Option<String>,
pub risk_per_trade: String,
pub max_daily_loss: String,
} }
+173 -90
View File
@@ -1,7 +1,6 @@
use crate::{engine::Engine, store::config::Config}; use crate::{engine::Engine, store::config::ConfigManager};
use pulse_sdk::prelude::*; use pulse_sdk::prelude::*;
use toml::Value;
impl Engine { impl Engine {
pub async fn invalid_command_usage(&self, name: &str) -> tokio::io::Result<()> { pub async fn invalid_command_usage(&self, name: &str) -> tokio::io::Result<()> {
@@ -13,7 +12,44 @@ impl Engine {
command: &str, command: &str,
mut args: Vec<&str>, mut args: Vec<&str>,
) -> tokio::io::Result<()> { ) -> 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 { 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 <name>", "Switch active account"),
("preset list", "List available presets"),
("preset use <name>", "Switch active preset"),
("strategy start", "Start strategy runtime"),
("strategy restart", "Restart strategy runtime"),
("strategy set <id>", "Change active strategy"),
("strategy <command>", "Send command to strategy runtime"),
];
for (command, description) in commands {
self.terminal_server
.info("engine::help", &format!("{:<30} {}", command, description))
.await?;
}
}
"config" | "cfg" => { "config" | "cfg" => {
if args.len() == 0 { if args.len() == 0 {
return self.invalid_command_usage("config").await; return self.invalid_command_usage("config").await;
@@ -21,10 +57,23 @@ impl Engine {
match args[0] { match args[0] {
"reload" => { "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 self.terminal_server
.info("engine::config", "successfully reloaded") .info("engine::config", "successfully reloaded")
.await?; .await?;
if strategy_changed {
self.strategy_engine
.reload(&config.get_ref()?.strategy)
.await?;
}
} }
"save" => { "save" => {
@@ -34,90 +83,6 @@ impl Engine {
.await?; .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?; self.invalid_command_usage("engine::config").await?;
} }
@@ -169,7 +134,7 @@ impl Engine {
.terminal_server .terminal_server
.error( .error(
"engine::account", "engine::account",
&format!("Account not found ({new_active})"), &format!("account not found ({new_active})"),
) )
.await; .await;
} }
@@ -179,7 +144,7 @@ impl Engine {
self.terminal_server self.terminal_server
.info( .info(
"engine::account", "engine::account",
&format!("Account set to {new_active} successfully!"), &format!("account set to {new_active} successfully!"),
) )
.await?; .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" => { "strategy" | "strat" | "sg" => {
let Some(strategy_command) = args.drain(0..=0).next() else { let Some(strategy_command) = args.drain(0..=0).next() else {
return self.invalid_command_usage("strategy").await; return self.invalid_command_usage("strategy").await;
@@ -198,6 +237,41 @@ impl Engine {
match strategy_command { match strategy_command {
"start" => {} "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 self.strategy_engine
.send(&StrategyEngineMessage::Command { .send(&StrategyEngineMessage::Command {
@@ -211,11 +285,20 @@ impl Engine {
_ => { _ => {
self.terminal_server self.terminal_server
.error("engine::cmd", &format!("Command '{}' not found", command)) .error(
"engine::cmd",
&format!("command '{}' not found, use help or ?", command),
)
.await?; .await?;
} }
} }
self.terminal_server
.broadcast(TerminalServerMessage::ConfigUpdated(
self.get_config_status().await?,
))
.await?;
Ok(()) Ok(())
} }
} }
+64 -38
View File
@@ -1,14 +1,24 @@
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 pulse_sdk::prelude::*;
use crate::engine::Engine; use crate::engine::Engine;
impl Engine { impl Engine {
pub async fn execute_signal(&self, signal: &Signal) -> tokio::io::Result<()> { pub async fn execute_signal(
&self,
size: &mut Option<EngineOrder>,
signal: &Signal,
) -> anyhow::Result<()> {
let client = hypercore::mainnet(); let client = hypercore::mainnet();
let accounts = self.accounts.lock().await; let accounts = self.accounts.lock().await;
if let Some(acc) = accounts.get_active() { let acc = accounts
.get_active()
.ok_or_else(|| anyhow::anyhow!("Unable to get active account"))?;
let Some(asset_id) = self let Some(asset_id) = self
.watch_list .watch_list
.lock() .lock()
@@ -17,80 +27,96 @@ impl Engine {
.get(&signal.symbol) .get(&signal.symbol)
.cloned() .cloned()
else { else {
self.terminal_server return Err(anyhow::anyhow!(
.error(
"self::order",
&format!(
"Invalid Symbol: {:?}, Unable to get asset id", "Invalid Symbol: {:?}, Unable to get asset id",
signal.symbol signal.symbol
), ));
)
.await?;
return Ok(());
}; };
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 { let order = BatchOrder {
orders: vec![ orders: vec![
OrderRequest { OrderRequest {
asset: asset_id, asset: asset_id,
is_buy: matches!(signal.side, Side::Bid), is_buy: matches!(signal.side, Side::Bid),
limit_px: signal.entry_price, limit_px,
sz: 0.into(), sz,
reduce_only: false, reduce_only: false,
order_type: OrderTypePlacement::Limit { order_type: OrderTypePlacement::Limit {
tif: TimeInForce::Gtc, tif: TimeInForce::Gtc,
}, },
cloid: Default::default(), cloid: engine_order.entry,
}, },
OrderRequest { OrderRequest {
asset: asset_id, asset: asset_id,
is_buy: matches!(signal.side, Side::Bid), is_buy: !matches!(signal.side, Side::Bid),
limit_px: signal.entry_price, limit_px,
sz: 0.into(), sz,
reduce_only: true, reduce_only: true,
order_type: OrderTypePlacement::Trigger { order_type: OrderTypePlacement::Trigger {
is_market: true, is_market: true,
trigger_px: signal.take_profit, trigger_px: round_price(signal.take_profit),
tpsl: hypercore::TpSl::Tp, tpsl: hypercore::TpSl::Tp,
}, },
cloid: Default::default(), cloid: engine_order.take_profit,
}, },
OrderRequest { OrderRequest {
asset: asset_id, asset: asset_id,
is_buy: matches!(signal.side, Side::Bid), is_buy: !matches!(signal.side, Side::Bid),
limit_px: signal.entry_price, limit_px,
sz: 0.into(), sz,
reduce_only: true, reduce_only: true,
order_type: OrderTypePlacement::Trigger { order_type: OrderTypePlacement::Trigger {
is_market: true, is_market: true,
trigger_px: signal.stop_loss, trigger_px: round_price(engine_order.stop_loss),
tpsl: hypercore::TpSl::Sl, tpsl: hypercore::TpSl::Sl,
}, },
cloid: Default::default(), cloid: engine_order.stop_loss_id,
}, },
], ],
grouping: hypercore::OrderGrouping::Na, grouping: hypercore::OrderGrouping::NormalTpsl,
builder: None, builder: None,
}; };
let nonce = chrono::Utc::now().timestamp_millis() as u64; println!("{order:?}");
let nonce = NonceHandler::default();
match client match client
.place(&acc.private_key.0, order, nonce, None, None) .place(&acc.private_key.0, order, nonce.next(), None, None)
.await .await
{ {
Ok(_) => {} // Invalid order
Err(e) => { Ok(o) if o.iter().any(|o| matches!(o, OrderResponseStatus::Error(_))) => {
self.terminal_server return Err(anyhow::anyhow!(
.error("self::order", &e.to_string()) "order rejected by HyperLiquid: {}",
.await?; o.into_iter()
} .enumerate()
} .filter_map(|(i, res)| {
if let OrderResponseStatus::Error(e) = res {
Some(format!("{i}={e}"))
} else { } else {
self.terminal_server None
.error("Engine::order", "Unable to get active account") }
.await?; })
.collect::<Vec<_>>()
.join(", ")
));
}
Ok(_) => {
self.risk_engine.order_placed(engine_order).await;
}
e => {
e?;
}
} }
Ok(()) Ok(())
+72 -16
View File
@@ -1,16 +1,17 @@
pub mod command; pub mod command;
pub mod execution; pub mod execution;
pub mod risk;
pub mod strategy; pub mod strategy;
pub mod terminal; pub mod terminal;
use crate::{ use crate::{
engine::{strategy::StrategyEngine, terminal::TerminalServer}, engine::{risk::RiskEngine, strategy::StrategyEngine, terminal::TerminalServer},
store::{accounts::AccountList, config::Config}, store::{accounts::AccountList, config::ConfigManager},
}; };
use hypersdk::hypercore::ws::ConnectionStream; use hypersdk::hypercore::ws::ConnectionStream;
use pulse_sdk::prelude::*; use pulse_sdk::prelude::*;
use std::{collections::HashMap, sync::Arc}; use std::{collections::HashMap, sync::Arc, time::Duration};
use tokio::sync::Mutex; use tokio::{sync::Mutex, time::Instant};
#[derive(Debug, Clone)] #[derive(Debug, Clone)]
pub struct WatchList { pub struct WatchList {
@@ -23,10 +24,10 @@ pub struct Engine {
// engine // engine
pub terminal_server: Arc<TerminalServer>, pub terminal_server: Arc<TerminalServer>,
pub strategy_engine: Arc<StrategyEngine>, pub strategy_engine: Arc<StrategyEngine>,
pub ws_stream: Arc<Mutex<ConnectionStream>>, pub risk_engine: Arc<RiskEngine>,
// data // data
pub config: Arc<Mutex<Config>>, pub config: Arc<Mutex<ConfigManager>>,
pub accounts: Arc<Mutex<AccountList>>, pub accounts: Arc<Mutex<AccountList>>,
// live data / status // live data / status
@@ -35,22 +36,29 @@ pub struct Engine {
pub signals: Arc<Mutex<Vec<SignalStatus>>>, pub signals: Arc<Mutex<Vec<SignalStatus>>>,
} }
pub struct EngineSides {
pub strategy_stream: ConnectionStream,
pub risk_stream: ConnectionStream,
}
impl Engine { impl Engine {
pub async fn new() -> tokio::io::Result<Arc<Self>> { pub async fn new() -> tokio::io::Result<(Arc<Self>, EngineSides)> {
let config = Config::new().await?; 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 accounts = Arc::new(Mutex::new(AccountList::new().await?));
let config = Arc::new(Mutex::new(config)); let config = Arc::new(Mutex::new(config));
Ok(Arc::new_cyclic(|engine| Self { Ok((
Arc::new_cyclic(|engine| Self {
// engine // engine
terminal_server: TerminalServer::new(engine.clone()), terminal_server: TerminalServer::new(engine.clone()),
strategy_engine: strategy.initialize(engine.clone()), strategy_engine: strategy.initialize(engine.clone()),
ws_stream: Arc::new(Mutex::new(ws_stream)), risk_engine: RiskEngine::new(engine.clone(), risk_handle),
// data // data
config, config,
@@ -58,15 +66,22 @@ impl Engine {
// live data / status // live data / status
status: Arc::new(Mutex::new(EngineStatus { status: Arc::new(Mutex::new(EngineStatus {
perps_equity: 0.into(),
strategy_mode: Mode::Auto, strategy_mode: Mode::Auto,
strategy_state: ItemState::Stopped, strategy_state: ItemState::Stopped,
positions: Vec::new(),
})), })),
watch_list: Arc::new(Mutex::new(WatchList { watch_list: Arc::new(Mutex::new(WatchList {
name_to_index: HashMap::new(), name_to_index: HashMap::new(),
items: Vec::new(), items: Vec::new(),
})), })),
signals: Arc::new(Mutex::new(Vec::new())), signals: Arc::new(Mutex::new(Vec::new())),
})) }),
EngineSides {
strategy_stream,
risk_stream,
},
))
} }
/// Starts the main strategy server /// Starts the main strategy server
@@ -89,12 +104,53 @@ impl Engine {
} }
} }
pub async fn get_config_status(&self) -> EngineConfig { pub async fn run_daily_scheduler(self: Arc<Self>) -> anyhow::Result<()> {
EngineConfig { loop {
strategy: self.strategy_engine.strategy.lock().await.manifest.clone(), 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<EngineConfig> {
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<()> { pub async fn update_status(&self) -> tokio::io::Result<()> {
self.terminal_server self.terminal_server
.broadcast(TerminalServerMessage::StatusUpdated( .broadcast(TerminalServerMessage::StatusUpdated(
+182
View File
@@ -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<Vec<EngineOrder>>,
pub handle: Mutex<ConnectionHandle>,
pub state: Mutex<RiskState>,
pub engine: Weak<Engine>,
}
impl RiskEngine {
pub fn new(engine: Weak<Engine>, handle: ConnectionHandle) -> Arc<Self> {
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<Self>,
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<bool> {
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<EngineOrder> {
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<RiskConfig> {
Ok(self.get_engine().config.lock().await.get_ref()?.risk)
}
pub fn get_engine(&self) -> Arc<Engine> {
self.engine
.upgrade()
.expect("Failed to upgrade engine(Weak) to Arc")
}
}
+68 -13
View File
@@ -1,6 +1,11 @@
use anyhow::Context;
use hypersdk::hypercore::{self, CandleInterval, Subscription, ws::ConnectionHandle};
use pulse_sdk::prelude::*; use pulse_sdk::prelude::*;
use anyhow::Context;
use futures::StreamExt;
use hypersdk::hypercore::{
self, CandleInterval, Subscription,
ws::{ConnectionHandle, ConnectionStream, Event},
};
use std::{ use std::{
collections::HashSet, collections::HashSet,
sync::{Arc, Weak}, sync::{Arc, Weak},
@@ -75,11 +80,40 @@ impl StrategyEngine {
Ok(()) Ok(())
} }
pub async fn run_event_stream(
self: Arc<Self>,
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<Self>) -> anyhow::Result<()> { pub async fn run(self: &Arc<Self>) -> anyhow::Result<()> {
let engine = self let engine = self.get_engine();
.engine
.upgrade()
.expect("Failed to upgrade engine (StrategyEngine)");
let mut stdout = { let mut stdout = {
let mut child = self.strategy.lock().await; let mut child = self.strategy.lock().await;
@@ -117,18 +151,33 @@ impl StrategyEngine {
engine.terminal_server.log_raw(log).await?; engine.terminal_server.log_raw(log).await?;
} }
Some(StrategyMessage::Signal(signal)) => { Some(StrategyMessage::Signal(mut signal)) => {
match engine.execute_signal(&signal).await { if !engine.risk_engine.validate_signal(&mut signal).await? {
Ok(_) => engine.signals.lock().await.push(Ok(signal)), engine.signals.lock().await.push(Err((signal, None)));
Err(e) => {
engine.signals.lock().await.push(Err(signal));
engine engine
.terminal_server .terminal_server
.error("signal", &format!("Failed to execute signal: {e}")) .error("engine::risk", "Signal rejected")
.await? .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("engine::order", &format!("Failed to execute signal: {e}"))
.await?;
Err((signal, size))
} }
};
engine.signals.lock().await.push(result);
engine engine
.terminal_server .terminal_server
@@ -197,4 +246,10 @@ impl StrategyEngine {
} }
} }
} }
pub fn get_engine(&self) -> Arc<Engine> {
self.engine
.upgrade()
.expect("Failed to upgrade engine (StrategyEngine)")
}
} }
+27 -14
View File
@@ -1,5 +1,6 @@
use crate::engine::Engine;
use pulse_sdk::{map_postcard_err, prelude::*}; use pulse_sdk::{map_postcard_err, prelude::*};
use crate::engine::Engine;
use std::{ use std::{
collections::HashMap, collections::HashMap,
sync::{Arc, Weak}, sync::{Arc, Weak},
@@ -37,7 +38,7 @@ impl TerminalServer {
} }
pub async fn run_server(self: Arc<Self>) -> tokio::io::Result<()> { pub async fn run_server(self: Arc<Self>) -> tokio::io::Result<()> {
let path = pulse_sdk::server_path(); let path = server_path();
if path.exists() { if path.exists() {
tokio::fs::remove_file(&path).await?; tokio::fs::remove_file(&path).await?;
@@ -91,7 +92,7 @@ impl TerminalServer {
self.send_to( self.send_to(
id, id,
TerminalServerMessage::ConfigUpdated(engine.get_config_status().await), TerminalServerMessage::ConfigUpdated(engine.get_config_status().await?),
) )
.await?; .await?;
@@ -103,7 +104,13 @@ impl TerminalServer {
self.send_to( self.send_to(
id, 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?; .await?;
@@ -151,7 +158,7 @@ impl TerminalServer {
pub async fn broadcast( pub async fn broadcast(
self: &Arc<Self>, self: &Arc<Self>,
message: pulse_sdk::terminal::TerminalServerMessage, message: TerminalServerMessage,
) -> tokio::io::Result<()> { ) -> tokio::io::Result<()> {
let msg = map_postcard_err(postcard::to_allocvec(&message))?; let msg = map_postcard_err(postcard::to_allocvec(&message))?;
@@ -194,7 +201,10 @@ impl TerminalServer {
loop { loop {
refresh.tick().await; refresh.tick().await;
match crate::fetch::fetch_watch_list(&client, &engine.config.lock().await.watchlist) match crate::fetch::fetch_watch_list(
&client,
&engine.config.lock().await.get_ref()?.watchlist,
)
.await .await
{ {
Ok(watch_list) => { Ok(watch_list) => {
@@ -224,8 +234,10 @@ impl TerminalServer {
if let Some(acc) = engine.accounts.lock().await.get_active() { if let Some(acc) = engine.accounts.lock().await.get_active() {
match client.clearinghouse_state(acc.address, None).await { match client.clearinghouse_state(acc.address, None).await {
Ok(state) => { Ok(state) => {
self.broadcast(TerminalServerMessage::PositionsUpdated( let mut status = engine.status.lock().await;
state
status.perps_equity = state.margin_summary.account_value;
status.positions = state
.asset_positions .asset_positions
.into_iter() .into_iter()
.map(|position| Position { .map(|position| Position {
@@ -234,9 +246,11 @@ impl TerminalServer {
entry_price: position.position.entry_px.unwrap_or_default(), entry_price: position.position.entry_px.unwrap_or_default(),
pnl: position.position.unrealized_pnl, pnl: position.position.unrealized_pnl,
}) })
.collect(), .collect();
))
.await?; drop(status);
engine.update_status().await?;
} }
Err(e) => { Err(e) => {
self.error("orders", &format!("Unable to get open orders: {e}")) self.error("orders", &format!("Unable to get open orders: {e}"))
@@ -256,7 +270,7 @@ impl TerminalServer {
pub async fn send_to( pub async fn send_to(
self: &Arc<Self>, self: &Arc<Self>,
id: &usize, id: &usize,
message: pulse_sdk::terminal::TerminalServerMessage, message: TerminalServerMessage,
) -> tokio::io::Result<()> { ) -> tokio::io::Result<()> {
Self::send_to_client( Self::send_to_client(
self.clients.lock().await.get_mut(id).ok_or_else(|| { self.clients.lock().await.get_mut(id).ok_or_else(|| {
@@ -312,8 +326,7 @@ impl TerminalServer {
self.logs.lock().await.push(log.clone()); self.logs.lock().await.push(log.clone());
self.broadcast(pulse_sdk::terminal::TerminalServerMessage::AddLog(log)) self.broadcast(TerminalServerMessage::AddLog(log)).await
.await
} }
pub async fn info(self: &Arc<Self>, name: &str, message: &str) -> tokio::io::Result<()> { pub async fn info(self: &Arc<Self>, name: &str, message: &str) -> tokio::io::Result<()> {
+24 -3
View File
@@ -4,15 +4,36 @@ pub mod store;
#[tokio::main] #[tokio::main]
async fn main() -> anyhow::Result<()> { 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 server = engine.terminal_server.spawn_server().await;
let broadcaster = engine.terminal_server.spawn_broadcaster().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?; engine.run().await?;
server.await??; server.abort();
broadcaster.await??; broadcaster.abort();
risk.abort();
strategy.abort();
daily_scheduler.abort();
Ok(()) Ok(())
} }
+62 -4
View File
@@ -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)] #[derive(Debug, serde::Serialize, serde::Deserialize)]
pub struct Config { pub struct Config {
pub description: Option<String>,
pub watchlist: Vec<String>, pub watchlist: Vec<String>,
pub strategy: String, 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<String, Config>,
}
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 { impl Default for Config {
fn default() -> Self { fn default() -> Self {
Self { Self {
description: None,
watchlist: vec!["BTC".to_string(), "SOL".to_string(), "ETH".to_string()], watchlist: vec!["BTC".to_string(), "SOL".to_string(), "ETH".to_string()],
strategy: String::new(), 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<Self> { pub async fn new() -> tokio::io::Result<Self> {
let path = crate::store::pulse_config_file()?; let path = crate::store::pulse_config_file()?;
+2 -5
View File
@@ -1,9 +1,6 @@
use std::{path::PathBuf, process::Stdio}; use pulse_sdk::{map_postcard_err, prelude::*};
use pulse_sdk::{ use std::{path::PathBuf, process::Stdio};
map_postcard_err,
strategy::{StrategyEngineMessage, StrategyManifest, StrategyMessage},
};
use tokio::{ use tokio::{
io::{AsyncReadExt, AsyncWriteExt}, io::{AsyncReadExt, AsyncWriteExt},
process::{Child, ChildStdout, Command}, process::{Child, ChildStdout, Command},
+45 -26
View File
@@ -1,6 +1,7 @@
use hypersdk::hypercore::Side;
use pulse_sdk::prelude::*; use pulse_sdk::prelude::*;
use hypersdk::hypercore::Side;
pub trait Formatted { pub trait Formatted {
fn get_formatted(&self) -> Vec<String>; fn get_formatted(&self) -> Vec<String>;
} }
@@ -54,32 +55,28 @@ impl Formatted for EventLog {
impl Formatted for SignalStatus { impl Formatted for SignalStatus {
fn get_formatted(&self) -> Vec<String> { fn get_formatted(&self) -> Vec<String> {
match self { match self {
Ok(signal) => { Ok((signal, order)) | Err((signal, order)) => {
vec![ let mut base = vec![
format!("\x1b[33mOK\x1b[0m"), if matches!(self, Ok(_)) {
if matches!(signal.side, Side::Ask) { format!("\x1b[34mOK\x1b[0m")
} else {
format!("\x1b[31mREJ\x1b[0m")
},
if matches!(signal.side, Side::Bid) {
format!("\x1b[32mBUY\x1b[0m") format!("\x1b[32mBUY\x1b[0m")
} else { } else {
format!("\x1b[31mSELL\x1b[0m") format!("\x1b[31mSELL\x1b[0m")
}, },
format_symbol(&signal.symbol), format_symbol(&signal.symbol),
format_usd(signal.entry_price.as_f64()), format_usd(signal.entry_price.as_f64()),
format!("\x1b[33mAPR {}\x1b[0m", signal.confidence), format_usd(signal.take_profit.as_f64()),
] ];
if let Some(order) = order {
base.extend(vec![format_usd_reverse(order.stop_loss.as_f64())]);
} }
Err(signal) => { base
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),
]
} }
} }
} }
@@ -199,14 +196,14 @@ impl Formatted for EngineStatus {
fn get_formatted(&self) -> Vec<String> { fn get_formatted(&self) -> Vec<String> {
vec![ vec![
Triple( Triple(
"\x1b[2mState\x1b[0m", "\x1b[2mPerps Equity\x1b[0m",
"\x1b[2mMode\x1b[0m", "\x1b[2mStrategy State\x1b[0m",
"\x1b[2m..\x1b[0m", "\x1b[2mStrategy Mode\x1b[0m",
), ),
Triple( Triple(
&format_usd(self.perps_equity.as_f64()),
&self.strategy_state.to_string(), &self.strategy_state.to_string(),
&self.strategy_mode.to_string(), &self.strategy_mode.to_string(),
"..",
), ),
] ]
.get_formatted() .get_formatted()
@@ -217,11 +214,25 @@ impl Formatted for EngineConfig {
fn get_formatted(&self) -> Vec<String> { fn get_formatted(&self) -> Vec<String> {
vec![ vec![
Triple( Triple(
"\x1b[2mPreset\x1b[0m",
"\x1b[2mStrategy\x1b[0m", "\x1b[2mStrategy\x1b[0m",
"\x1b[2m..\x1b[0m", "\x1b[2mDescription\x1b[0m",
"\x1b[2m..\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() .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 { pub fn format_f64(value: f64) -> String {
let abs = value.abs(); let abs = value.abs();
+26 -5
View File
@@ -2,6 +2,8 @@ pub mod command;
pub mod formatting; pub mod formatting;
pub mod terminal; pub mod terminal;
use pulse_sdk::prelude::*;
use std::any::Any; use std::any::Any;
use chrono::{Local, Utc}; use chrono::{Local, Utc};
@@ -21,8 +23,6 @@ use pulse_ui::{
use crate::formatting::{Formatted, apply_padding}; use crate::formatting::{Formatted, apply_padding};
use pulse_sdk::prelude::*;
pub struct PulseTradeApp { pub struct PulseTradeApp {
sock: Option<terminal::TerminalClient>, sock: Option<terminal::TerminalClient>,
@@ -33,7 +33,6 @@ pub struct PulseTradeApp {
status: State<Option<EngineStatus>>, status: State<Option<EngineStatus>>,
watch_list: State<Vec<MarketItem>>, watch_list: State<Vec<MarketItem>>,
active_positions: State<Vec<Position>>,
logs: State<Vec<EventLog>>, logs: State<Vec<EventLog>>,
signals: State<Vec<SignalStatus>>, signals: State<Vec<SignalStatus>>,
inspect: State<InspectTarget>, inspect: State<InspectTarget>,
@@ -128,7 +127,10 @@ impl App for PulseTradeApp {
( (
LayoutItem::Widget(Size::Flex(1)), LayoutItem::Widget(Size::Flex(1)),
Box::new( 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()), command: ctx.use_state(InputState::new()),
scroll: ctx.use_state(ScrollState(1, [0; 7])), scroll: ctx.use_state(ScrollState(1, [0; 7])),
watch_list: ctx.use_state(Vec::new()), watch_list: ctx.use_state(Vec::new()),
active_positions: ctx.use_state(Vec::new()),
signals: ctx.use_state(Vec::new()), signals: ctx.use_state(Vec::new()),
logs: ctx.use_state(Vec::new()), logs: ctx.use_state(Vec::new()),
inspect: ctx.use_state(InspectTarget::None), inspect: ctx.use_state(InspectTarget::None),
@@ -190,6 +191,26 @@ async fn main() -> tokio::io::Result<()> {
Ok(()) Ok(())
} }
pub async fn advanced_draw_map<const N: usize, T, B: Formatted>(
scroll: &State<ScrollState<N>>,
index: usize,
title: &'static str,
state: &State<Option<T>>,
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<const N: usize, T: Formatted>( pub async fn advanced_draw<const N: usize, T: Formatted>(
scroll: &State<ScrollState<N>>, scroll: &State<ScrollState<N>>,
index: usize, index: usize,
+22 -8
View File
@@ -1,4 +1,5 @@
use pulse_sdk::{map_postcard_err, prelude::*}; use pulse_sdk::{map_postcard_err, prelude::*};
use pulse_ui::state::State; use pulse_ui::state::State;
use tokio::{ use tokio::{
@@ -46,7 +47,6 @@ impl TerminalClient {
let reader = reader.expect("Reader failed to swap"); let reader = reader.expect("Reader failed to swap");
let watch_list = app.watch_list.clone(); let watch_list = app.watch_list.clone();
let active_positions = app.active_positions.clone();
let logs = app.logs.clone(); let logs = app.logs.clone();
let signals = app.signals.clone(); let signals = app.signals.clone();
let market_overview = app.config.clone(); let market_overview = app.config.clone();
@@ -57,7 +57,6 @@ impl TerminalClient {
if let Err(v) = Self::run_client( if let Err(v) = Self::run_client(
reader, reader,
watch_list, watch_list,
active_positions,
logs, logs,
signals, signals,
market_overview, market_overview,
@@ -66,7 +65,27 @@ impl TerminalClient {
) )
.await .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, mut reader: OwnedReadHalf,
watch_list: State<Vec<MarketItem>>, watch_list: State<Vec<MarketItem>>,
active_positions: State<Vec<Position>>,
logs: State<Vec<EventLog>>, logs: State<Vec<EventLog>>,
signals: State<Vec<SignalStatus>>, signals: State<Vec<SignalStatus>>,
config: State<Option<EngineConfig>>, config: State<Option<EngineConfig>>,
@@ -107,10 +125,6 @@ impl TerminalClient {
*watch_list.lock().await = v; *watch_list.lock().await = v;
} }
TerminalServerMessage::PositionsUpdated(v) => {
*active_positions.lock().await = v;
}
TerminalServerMessage::ConfigUpdated(v) => { TerminalServerMessage::ConfigUpdated(v) => {
*config.lock().await = Some(v); *config.lock().await = Some(v);
} }