price changer refactor

Cargo.toml

@@ -19,4 +19,5 @@ clap = { version = "4.4.8", features = ["derive"] }
 serde_json = "1.0.113"
 log = "0.4.20"
 futures-util = "0.3.30"
-RustQuant = "0.0.49"
+RustQuant = "0.0.49"
+futures = "0.3.30"

configs/stable_pool.toml

@@ -8,4 +8,4 @@ PoolAdmin = {}
 TokenAdmin = { token_data.TKN0 = { name = "TKN0", symbol = "Token 0", decimals = 18 } }
 
 [[changer]]
-PriceChanger = { mu = 1, theta = 1, sigma = 1, cursor = 0, value = 1 }
+PriceChanger = { mu = 1, theta = 1, sigma = 1, cursor = 0, value = 1, seed = 69 }

src/behaviors/deployer.rs

@@ -15,8 +15,8 @@ use crate::bindings::uniswap_v3_factory::UniswapV3Factory;
 
 #[derive(Debug, Deserialize, Serialize)]
 pub struct DeploymentData {
-    factory: H160,
-    liquid_exchange: H160,
+    pub factory: H160,
+    pub liquid_exchange: H160,
 }
 
 impl DeploymentData {

src/behaviors/pool_admin.rs

@@ -42,7 +42,7 @@ impl Behavior<Message> for PoolAdmin {
         self.client = Some(client.clone());
         self.messager = Some(messager.clone());
 
-        Ok(None)
+        Ok(Some(messager.clone().stream().unwrap()))
     }
 
     async fn process(&mut self, event: Message) -> Result<ControlFlow> {
@@ -172,7 +172,7 @@ mod tests {
         world.add_agent(pool_admin);
         world.add_agent(deployer);
 
-        world.run().await.expect("world failed to run");
+        let _ = world.run().await.expect("world failed to run");
 
         let env = world.environment;
         let mut world = World::new("univ3");

src/behaviors/price_changer.rs

@@ -4,27 +4,32 @@ use anyhow::Result;
 use arbiter_core::middleware::ArbiterMiddleware;
 use arbiter_engine::messager::{Message, Messager};
 use ethers::types::H160;
+use futures::stream::StreamExt;
 use RustQuant::{
     models::*,
     stochastics::{process::Trajectories, *},
 };
 
 use super::*;
-use crate::bindings::liquid_exchange::LiquidExchange;
+use crate::{behaviors::deployer::DeploymentData, bindings::liquid_exchange::LiquidExchange};
 
 #[derive(Serialize, Deserialize)]
 pub struct PriceChanger {
     mu: f64,
     sigma: f64,
     theta: f64,
 
+    seed: u64,
+
     #[serde(skip)]
     #[serde(default = "trajectory_default")]
     pub current_chunk: Trajectories,
 
     #[serde(skip)]
     pub client: Option<Arc<ArbiterMiddleware>>,
 
+    pub liquid_exchange: Option<H160>,
+
     cursor: usize,
     value: f64,
 }
@@ -43,25 +48,26 @@ impl fmt::Debug for PriceChanger {
 }
 
 #[derive(Debug, Serialize, Deserialize)]
-pub struct PriceUpdate {
-    liquid_exchange: H160,
-}
+pub struct PriceUpdate;
 
 impl PriceChanger {
     /// Public constructor function to create a [`PriceChanger`] behaviour.
-    pub fn new(initial_value: f64, mu: f64, sigma: f64, theta: f64) -> Self {
+    pub fn new(initial_value: f64, mu: f64, sigma: f64, theta: f64, seed: u64) -> Self {
         let ou = OrnsteinUhlenbeck::new(mu, sigma, theta);
 
         // Chunk our price trajectory over 100 price points.
-        let current_chunk = ou.euler_maruyama(initial_value, 0.0, 100.0, 100_usize, 1_usize, false);
+        let current_chunk =
+            ou.seedable_euler_maruyama(initial_value, 0.0, 100.0, 100_usize, 1_usize, false, seed);
 
         Self {
             mu,
             sigma,
             theta,
+            seed,
             current_chunk,
             cursor: 0,
             client: None,
+            liquid_exchange: None,
             value: initial_value,
         }
     }
@@ -72,17 +78,27 @@ impl Behavior<Message> for PriceChanger {
     async fn startup(
         &mut self,
         client: Arc<ArbiterMiddleware>,
-        _messager: Messager,
+        messager: Messager,
     ) -> Result<Option<EventStream<Message>>> {
         self.client = Some(client);
 
-        Ok(None)
+        while let Some(message) = messager.clone().stream().unwrap().next().await {
+            match serde_json::from_str::<DeploymentData>(&message.data) {
+                Ok(data) => {
+                    self.liquid_exchange = Some(data.liquid_exchange);
+                    break;
+                }
+                Err(_) => continue,
+            };
+        }
+
+        Ok(Some(messager.clone().stream().unwrap()))
     }
 
     async fn process(&mut self, event: Message) -> Result<ControlFlow> {
         let ou = OrnsteinUhlenbeck::new(self.mu, self.sigma, self.theta);
 
-        let query: PriceUpdate = match serde_json::from_str(&event.data) {
+        let _query: PriceUpdate = match serde_json::from_str(&event.data) {
             Ok(query) => query,
             Err(_) => {
                 eprintln!("Failed to deserialize the event data into a PriceUpdate");
@@ -93,12 +109,13 @@ impl Behavior<Message> for PriceChanger {
         if self.cursor >= 99 {
             self.cursor = 0;
             self.value = self.current_chunk.paths.clone()[0][self.cursor];
-            self.current_chunk =
-                ou.euler_maruyama(self.value, 0.0, 100.0, 100_usize, 1_usize, false);
+            self.current_chunk = ou.seedable_euler_maruyama(
+                self.value, 0.0, 100.0, 100_usize, 1_usize, false, self.seed,
+            );
         }
 
         let liquid_exchange =
-            LiquidExchange::new(query.liquid_exchange, self.client.clone().unwrap());
+            LiquidExchange::new(self.liquid_exchange.unwrap(), self.client.clone().unwrap());
 
         let price = self.current_chunk.paths.clone()[0][self.cursor];
 

src/behaviors/token_admin.rs

@@ -88,7 +88,7 @@ impl Behavior<Message> for TokenAdmin {
 
         let _ = messager.send(To::All, &message_content).await;
 
-        Ok(None)
+        Ok(Some(messager.clone().stream().unwrap()))
     }
 
     async fn process(&mut self, event: Message) -> Result<ControlFlow> {