diff --git a/src/engine/engine/mod.rs b/src/engine/engine/mod.rs index 9989ac9..a11422a 100644 --- a/src/engine/engine/mod.rs +++ b/src/engine/engine/mod.rs @@ -25,7 +25,6 @@ pub struct Engine { pub terminal_server: Arc, pub strategy_engine: Arc, pub risk_engine: Arc, - pub ws_stream: Arc>, // data pub config: Arc>, @@ -37,39 +36,50 @@ pub struct Engine { pub signals: Arc>>, } +pub struct EngineSides { + pub strategy_stream: ConnectionStream, + pub risk_stream: ConnectionStream, +} + impl Engine { - pub async fn new() -> tokio::io::Result> { + pub async fn new() -> tokio::io::Result<(Arc, EngineSides)> { 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.get_ref()?.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 config = Arc::new(Mutex::new(config)); - Ok(Arc::new_cyclic(|engine| Self { - // engine - terminal_server: TerminalServer::new(engine.clone()), - strategy_engine: strategy.initialize(engine.clone()), - risk_engine: RiskEngine::new(engine.clone()), - ws_stream: Arc::new(Mutex::new(ws_stream)), + Ok(( + Arc::new_cyclic(|engine| Self { + // engine + terminal_server: TerminalServer::new(engine.clone()), + strategy_engine: strategy.initialize(engine.clone()), + risk_engine: RiskEngine::new(engine.clone(), risk_handle), - // data - config, - accounts, + // data + config, + accounts, - // live data / status - status: Arc::new(Mutex::new(EngineStatus { - strategy_mode: Mode::Auto, - strategy_state: ItemState::Stopped, - })), - watch_list: Arc::new(Mutex::new(WatchList { - name_to_index: HashMap::new(), - items: Vec::new(), - })), - signals: Arc::new(Mutex::new(Vec::new())), - })) + // live data / status + status: Arc::new(Mutex::new(EngineStatus { + strategy_mode: Mode::Auto, + strategy_state: ItemState::Stopped, + })), + watch_list: Arc::new(Mutex::new(WatchList { + name_to_index: HashMap::new(), + items: Vec::new(), + })), + signals: Arc::new(Mutex::new(Vec::new())), + }), + EngineSides { + strategy_stream, + risk_stream, + }, + )) } /// Starts the main strategy server diff --git a/src/engine/engine/risk.rs b/src/engine/engine/risk.rs index a16e85d..58eb06d 100644 --- a/src/engine/engine/risk.rs +++ b/src/engine/engine/risk.rs @@ -1,6 +1,13 @@ use std::sync::{Arc, Weak}; -use hypersdk::hypercore::{self, Cloid, WebSocket}; +use futures::StreamExt; +use hypersdk::{ + Decimal, + hypercore::{ + self, Cloid, + ws::{ConnectionHandle, ConnectionStream}, + }, +}; use pulse_sdk::general::Signal; use tokio::sync::Mutex; @@ -23,22 +30,84 @@ impl OrderIds { } } +pub struct RiskState { + pub starting_equity: Decimal, + pub realized_pnl_today: Decimal, + pub open_positions: usize, +} + pub struct RiskEngine { // Orders made by the engine pub orders: Mutex>, - pub ws: Mutex, + pub handle: Mutex, + pub state: Mutex, pub engine: Weak, } impl RiskEngine { - pub fn new(engine: Weak) -> Arc { + pub fn new(engine: Weak, handle: ConnectionHandle) -> Arc { Arc::new(Self { orders: Mutex::new(Vec::new()), - ws: Mutex::new(hypercore::mainnet_ws()), + handle: Mutex::new(handle), + state: Mutex::new(RiskState { + starting_equity: 0.into(), + realized_pnl_today: 0.into(), + open_positions: 0, + }), engine, }) } + pub async fn day_tick(&self) -> anyhow::Result<()> { + let engine = self.get_engine(); + + let client = hypercore::mainnet(); + + if let Some(acc) = engine.accounts.lock().await.get_active() { + self.state.lock().await.starting_equity = client + .user_vault_equities(acc.address) + .await? + .into_iter() + .map(|v| v.equity) + .sum(); + } + + Ok(()) + } + + pub async fn run(&self, mut stream: ConnectionStream) -> anyhow::Result<()> { + while let Some(e) = stream.next().await { + match e { + hypercore::ws::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 == cloid { + if order.status.is_filled() { + rm.push(i); + } + + break; + } + } + + for r in rm { + self.orders.lock().await.remove(r); + } + } + } + } + _ => {} + } + } + + Ok(()) + } + pub async fn validate_signal(&self, _signal: &mut Signal) -> bool { true } diff --git a/src/engine/main.rs b/src/engine/main.rs index c9d3244..8531472 100644 --- a/src/engine/main.rs +++ b/src/engine/main.rs @@ -4,7 +4,7 @@ pub mod store; #[tokio::main] 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 broadcaster = engine.terminal_server.spawn_broadcaster().await;