diff --git a/src/engine/engine.rs b/src/engine/engine/mod.rs similarity index 98% rename from src/engine/engine.rs rename to src/engine/engine/mod.rs index 798ebb3..c510c6c 100644 --- a/src/engine/engine.rs +++ b/src/engine/engine/mod.rs @@ -1,5 +1,8 @@ +pub mod plugin; + use crate::{ - store::{accounts::AccountList, config::Config, plugin::StrategyEngine}, + engine::plugin::StrategyEngine, + store::{accounts::AccountList, config::Config}, terminal::TerminalServer, }; use pulse_wire::prelude::*; diff --git a/src/engine/engine/plugin.rs b/src/engine/engine/plugin.rs new file mode 100644 index 0000000..86545dd --- /dev/null +++ b/src/engine/engine/plugin.rs @@ -0,0 +1,174 @@ +use pulse_wire::prelude::*; +use std::{ + path::PathBuf, + process::Stdio, + sync::{Arc, Weak}, +}; +use tokio::{ + fs, + process::{Child, Command}, +}; + +use crate::{ + engine::Engine, + store::{plugin::Plugin, pulse_plugin}, +}; + +#[derive(Debug)] +pub struct StrategyEngine { + pub strategy: Arc>, + pub risk: Arc>, + + pub engine: Weak, +} + +impl StrategyEngine { + pub async fn new(strategy_id: &str, risk_id: &str) -> tokio::io::Result { + let strategy = pulse_plugin(strategy_id)?; + let risk = pulse_plugin(risk_id)?; + + let (strategy, strategy_manifest) = get_manifest_plugin_pair( + &strategy, + &strategy.join("strategy.bash"), + &fs::read(strategy.join("strategy.toml")).await?, + )?; + + let (risk, risk_manifest) = get_manifest_plugin_pair( + &risk, + &risk.join("risk.bash"), + &fs::read(risk.join("risk.toml")).await?, + )?; + + Ok(Self { + strategy: Arc::new(Plugin::new(strategy, strategy_manifest)), + risk: Arc::new(Plugin::new(risk, risk_manifest)), + engine: Weak::new(), + }) + } + + pub fn initialize(mut self, engine: Weak) -> Arc { + self.engine = engine; + + Arc::new(self) + } + + pub async fn run_strategy(&self) -> tokio::io::Result<()> { + let engine = self + .engine + .upgrade() + .expect("Failed to upgrade engine (StrategyEngine)"); + + let strategy = self.strategy.clone(); + let risk = self.risk.clone(); + + loop { + match strategy.recv().await? { + None => {} + + Some(StrategyMessage::Log(mut log)) => { + log.name.insert_str(0, "strategy::"); + engine.terminal_server.log_raw(log).await?; + } + + Some(StrategyMessage::Signal(signal)) => { + risk.send(&RiskEngineMessage::Signal(signal)).await?; + } + + Some(StrategyMessage::SubscribeCandle { symbol, timeframe }) => {} + + Some(StrategyMessage::RequestOHLC { + symbol, + timeframe, + count, + }) => {} + } + } + } + + pub async fn run_risk(&self) -> tokio::io::Result<()> { + let engine = self + .engine + .upgrade() + .expect("Failed to upgrade engine (StrategyEngine)"); + + let risk = self.risk.clone(); + + loop { + match risk.recv().await? { + None => {} + + Some(RiskMessage::Log(mut log)) => { + log.name.insert_str(0, "strategy::"); + engine.terminal_server.log_raw(log).await?; + } + + Some(RiskMessage::Approve(signal)) => {} + Some(RiskMessage::Reject { reason }) => {} + } + } + + Ok(()) + } + + pub async fn spawn(self: &Arc) { + let engine = self.clone(); + + tokio::spawn(async move { engine.run_strategy().await }); + + let engine = self.clone(); + + tokio::spawn(async move { engine.run_risk().await }); + } + + pub async fn reload_strategy(self: &Arc, id: &str) -> tokio::io::Result<()> { + let plugin = pulse_plugin(id)?; + + let (child, manifest) = get_manifest_plugin_pair( + &plugin, + &plugin.join("strategy.bash"), + &fs::read(plugin.join("strategy.toml")).await?, + )?; + + self.strategy.reload(child, manifest).await?; + + let engine = self.clone(); + tokio::spawn(async move { engine.run_strategy().await }); + + Ok(()) + } + + pub async fn reload_risk(self: &Arc, id: &str) -> tokio::io::Result<()> { + let plugin = pulse_plugin(id)?; + + let (child, manifest) = get_manifest_plugin_pair( + &plugin, + &plugin.join("risk.bash"), + &fs::read(plugin.join("risk.toml")).await?, + )?; + + self.risk.reload(child, manifest).await?; + + let engine = self.clone(); + tokio::spawn(async move { engine.run_risk().await }); + + Ok(()) + } +} + +pub fn get_manifest_plugin_pair<'de, M: serde::Deserialize<'de>>( + plugin_dir: &PathBuf, + plugin_path: &PathBuf, + manifest: &'de [u8], +) -> tokio::io::Result<(Child, M)> { + Ok(( + Command::new("bash") + .arg(plugin_path) + .current_dir(plugin_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()))?, + )) +} diff --git a/src/engine/store/plugin.rs b/src/engine/store/plugin.rs index 3a83bd9..6f37cb4 100644 --- a/src/engine/store/plugin.rs +++ b/src/engine/store/plugin.rs @@ -1,28 +1,14 @@ -use std::{ - marker::PhantomData, - path::PathBuf, - process::Stdio, - sync::{Arc, Weak}, -}; +use std::marker::PhantomData; -use pulse_wire::{ - PulseWire, - plugin::{ - RiskEngineMessage, RiskManifest, RiskMessage, StrategyEngineMessage, StrategyManifest, - StrategyMessage, - }, -}; +use pulse_wire::PulseWire; use serde::Deserialize; use tokio::{ - fs, io::{AsyncReadExt, AsyncWriteExt}, - process::{Child, ChildStdout, Command}, + process::{Child, ChildStdout}, sync::Mutex, }; -use crate::{engine::Engine, store::pulse_plugin}; - #[derive(Debug)] pub struct Plugin Deserialize<'de>> { pub manifest: Mutex, @@ -89,181 +75,3 @@ impl Deserialize<'de>> Plugin { Ok(()) } } - -pub fn get_manifest_plugin_pair<'de, M: Deserialize<'de>>( - plugin_dir: &PathBuf, - plugin_path: &PathBuf, - manifest: &'de [u8], -) -> tokio::io::Result<(Child, M)> { - Ok(( - Command::new("bash") - .arg(plugin_path) - .current_dir(plugin_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()))?, - )) -} - -#[derive(Debug)] -pub struct StrategyPair { - pub strategy: Plugin, - pub risk: Plugin, -} - -impl StrategyPair { - pub async fn new(strategy_id: &str, risk_id: &str) -> tokio::io::Result> { - let strategy = pulse_plugin(strategy_id)?; - let risk = pulse_plugin(risk_id)?; - - let (strategy, strategy_manifest) = get_manifest_plugin_pair( - &strategy, - &strategy.join("strategy.bash"), - &fs::read(strategy.join("strategy.toml")).await?, - )?; - - let (risk, risk_manifest) = get_manifest_plugin_pair( - &risk, - &risk.join("risk.bash"), - &fs::read(risk.join("risk.toml")).await?, - )?; - - Ok(Arc::new(Self { - strategy: Plugin::new(strategy, strategy_manifest), - risk: Plugin::new(risk, risk_manifest), - })) - } -} - -#[derive(Debug)] -pub struct StrategyEngine { - pub pair: Mutex>, - pub engine: Weak, -} - -impl StrategyEngine { - pub async fn new(strategy_id: &str, risk_id: &str) -> tokio::io::Result { - Ok(Self { - pair: Mutex::new(StrategyPair::new(strategy_id, risk_id).await?), - engine: Weak::new(), - }) - } - - pub fn initialize(mut self, engine: Weak) -> Arc { - self.engine = engine; - - Arc::new(self) - } - - pub async fn run_strategy(&self) -> tokio::io::Result<()> { - let engine = self - .engine - .upgrade() - .expect("Failed to upgrade engine (StrategyEngine)"); - - let pair = self.pair.lock().await.clone(); - - loop { - match pair.strategy.recv().await? { - None => {} - - Some(StrategyMessage::Log(mut log)) => { - log.name.insert_str(0, "strategy::"); - engine.terminal_server.log_raw(log).await?; - } - - Some(StrategyMessage::Signal(signal)) => { - pair.risk.send(&RiskEngineMessage::Signal(signal)).await?; - } - - Some(StrategyMessage::SubscribeCandle { symbol, timeframe }) => {} - - Some(StrategyMessage::RequestOHLC { - symbol, - timeframe, - count, - }) => {} - } - } - } - - pub async fn run_risk(&self) -> tokio::io::Result<()> { - let engine = self - .engine - .upgrade() - .expect("Failed to upgrade engine (StrategyEngine)"); - - let pair = self.pair.lock().await.clone(); - - loop { - match pair.risk.recv().await? { - None => {} - - Some(RiskMessage::Log(mut log)) => { - log.name.insert_str(0, "strategy::"); - engine.terminal_server.log_raw(log).await?; - } - - Some(RiskMessage::Approve(signal)) => {} - Some(RiskMessage::Reject { reason }) => {} - } - } - - Ok(()) - } - - pub async fn spawn(self: &Arc) { - let engine = self.clone(); - - tokio::spawn(async move { engine.run_strategy().await }); - - let engine = self.clone(); - - tokio::spawn(async move { engine.run_risk().await }); - } - - pub async fn reload_strategy(self: &Arc, id: &str) -> tokio::io::Result<()> { - { - let pair = self.pair.lock().await; - - let plugin = pulse_plugin(id)?; - - let (child, manifest) = get_manifest_plugin_pair( - &plugin, - &plugin.join("strategy.bash"), - &fs::read(plugin.join("strategy.toml")).await?, - )?; - - pair.strategy.reload(child, manifest).await?; - } - - let engine = self.clone(); - tokio::spawn(async move { engine.run_strategy().await }); - - Ok(()) - } - - pub async fn reload_risk(self: &Arc, id: &str) -> tokio::io::Result<()> { - { - let pair = self.pair.lock().await; - - let plugin = pulse_plugin(id)?; - - let (child, manifest) = get_manifest_plugin_pair( - &plugin, - &plugin.join("risk.bash"), - &fs::read(plugin.join("risk.toml")).await?, - )?; - - pair.risk.reload(child, manifest).await?; - } - - let engine = self.clone(); - tokio::spawn(async move { engine.run_risk().await }); - - Ok(()) - } -} diff --git a/src/engine/terminal.rs b/src/engine/terminal.rs index 13aec15..f76c9d2 100644 --- a/src/engine/terminal.rs +++ b/src/engine/terminal.rs @@ -61,13 +61,12 @@ impl TerminalServer { async fn initialize_client(self: &Arc, id: &usize) -> tokio::io::Result<()> { let engine = self.get_engine(); - let pair = engine.strategy.pair.lock().await; self.send_to( id, pulse_wire::terminal::TerminalServerMessage::StrategyUpdated(Strategy { - strategy: pair.strategy.manifest.lock().await.clone(), - risk: pair.risk.manifest.lock().await.clone(), + strategy: engine.strategy.strategy.manifest.lock().await.clone(), + risk: engine.strategy.risk.manifest.lock().await.clone(), mode: Mode::Auto, state: ItemState::Running, cooldown: TimeFrame::M15,