feat(bot): implement main orchestrator with multi-tier system and WS/REST server

Makefile

@@ -56,8 +56,7 @@ db: ## Connect to the Postgres database using psql locally
 db-shell: ## Open psql inside the postgres container
 	docker compose exec postgres psql -U postgres -d arbos
 
-redis-cli: ## Open redis-cli inside the redis container
-	docker compose exec redis redis-cli
+
 
 # ── Testing ───────────────────────────────────────────────────────────────────
 

crates/bot/Cargo.toml

@@ -6,5 +6,19 @@ edition = "2024"
 [dependencies]
 arbos-core = { path = "../core" }
 arbos-engine = { path = "../engine" }
+arbos-ingestor = { path = "../ingestor" }
 tokio = { workspace = true }
 polymarket-client-sdk = { workspace = true }
+tracing = { workspace = true }
+tracing-subscriber = { workspace = true }
+anyhow = { workspace = true }
+serde = { workspace = true }
+serde_json = { workspace = true }
+rust_decimal = { version = "1.40.0", features = ["macros"] }
+axum = { version = "0.8", features = ["ws"] }
+tower-http = { version = "0.6", features = ["cors"] }
+tokio-util = { version = "0.7.18", features = ["rt"] }
+
+[dev-dependencies]
+rust_decimal_macros = "1.40.0"
+tower = "0.5.3"

crates/bot/src/config.rs

@@ -0,0 +1,117 @@
+use arbos_core::domain::ExecutionMode;
+use polymarket_client_sdk::types::U256;
+use std::str::FromStr;
+use tracing::warn;
+
+/// Centralized configuration for the Bot orchestrator.
+/// Parsed from environment variables with sensible defaults.
+pub struct BotConfig {
+    pub execution_mode: ExecutionMode,
+    pub initial_assets: Vec<U256>,
+    pub ws_port: u16,
+    pub dedup_cooldown_secs: u64,
+    pub max_history_entries: usize,
+}
+
+impl BotConfig {
+    /// Parse configuration from environment variables.
+    ///
+    /// | Variable               | Default        | Description                              |
+    /// |------------------------|----------------|------------------------------------------|
+    /// | `ARBOS_LIVE_MODE`      | `false`        | Enable live trading (not yet implemented) |
+    /// | `ARBOS_INITIAL_ASSETS` | (none)         | Comma-separated initial asset IDs        |
+    /// | `ARBOS_WS_PORT`        | `3001`         | WebSocket server port for frontend       |
+    /// | `ARBOS_DEDUP_COOLDOWN` | `30`           | Signal dedup cooldown in seconds         |
+    /// | `ARBOS_MAX_HISTORY`    | `500`          | Max execution reports to keep in memory  |
+    pub fn from_env() -> Self {
+        let execution_mode = if std::env::var("ARBOS_LIVE_MODE")
+            .unwrap_or_default()
+            .eq_ignore_ascii_case("true")
+        {
+            warn!("ARBOS_LIVE_MODE=true detected. Live execution is NOT YET IMPLEMENTED.");
+            ExecutionMode::Live
+        } else {
+            ExecutionMode::Demo
+        };
+
+        let initial_assets: Vec<U256> = std::env::var("ARBOS_INITIAL_ASSETS")
+            .unwrap_or_default()
+            .split(',')
+            .filter(|s| !s.trim().is_empty())
+            .filter_map(|s| {
+                U256::from_str(s.trim())
+                    .inspect_err(|e| {
+                        warn!(asset = s.trim(), error = %e, "Skipping invalid asset ID");
+                    })
+                    .ok()
+            })
+            .collect();
+
+        let ws_port = std::env::var("ARBOS_WS_PORT")
+            .ok()
+            .and_then(|v| v.parse().ok())
+            .unwrap_or(3001);
+
+        let dedup_cooldown_secs = std::env::var("ARBOS_DEDUP_COOLDOWN")
+            .ok()
+            .and_then(|v| v.parse().ok())
+            .unwrap_or(30);
+
+        let max_history_entries = std::env::var("ARBOS_MAX_HISTORY")
+            .ok()
+            .and_then(|v| v.parse().ok())
+            .unwrap_or(500);
+
+        Self {
+            execution_mode,
+            initial_assets,
+            ws_port,
+            dedup_cooldown_secs,
+            max_history_entries,
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_config_defaults() {
+        // Run test in an isolated env space (or temporarily clear the relevant variables)
+        // to avoid flakiness in CI environments that set some ARBOS_ vars.
+        let vars_to_clear = [
+            "ARBOS_LIVE_MODE",
+            "ARBOS_INITIAL_ASSETS",
+            "ARBOS_WS_PORT",
+            "ARBOS_DEDUP_COOLDOWN",
+            "ARBOS_MAX_HISTORY",
+        ];
+
+        // Save current values
+        let mut saved_vars = Vec::new();
+        for var in &vars_to_clear {
+            saved_vars.push((*var, std::env::var(*var).ok()));
+            unsafe {
+                std::env::remove_var(*var);
+            }
+        }
+
+        // With no env vars set, should use safe defaults
+        let config = BotConfig::from_env();
+        assert_eq!(config.execution_mode, ExecutionMode::Demo);
+        assert!(config.initial_assets.is_empty());
+        assert_eq!(config.ws_port, 3001);
+        assert_eq!(config.dedup_cooldown_secs, 30);
+        assert_eq!(config.max_history_entries, 500);
+
+        // Restore prior env values
+        for (var, val_opt) in saved_vars {
+            if let Some(val) = val_opt {
+                unsafe {
+                    std::env::set_var(var, val);
+                }
+            }
+        }
+    }
+}

crates/bot/src/dedup.rs

@@ -0,0 +1,127 @@
+use polymarket_client_sdk::types::U256;
+use std::collections::HashMap;
+use std::hash::{DefaultHasher, Hash, Hasher};
+use std::time::{Duration, Instant};
+
+/// Prevents re-execution of the same arb pair within a configurable cooldown window.
+/// This guards against the Engine firing the same signal on consecutive book update ticks.
+pub struct SignalDeduplicator {
+    cooldown: Duration,
+    /// Maps a canonical hash key → the last time this combination of assets was executed.
+    recent_signals: HashMap<u64, Instant>,
+}
+
+impl SignalDeduplicator {
+    pub fn new(cooldown_secs: u64) -> Self {
+        Self {
+            cooldown: Duration::from_secs(cooldown_secs),
+            recent_signals: HashMap::new(),
+        }
+    }
+
+    /// Check if this pair/basket is allowed to execute. Returns `true` if the pair has not
+    /// been executed within the cooldown window. Automatically records the pair if allowed.
+    pub fn try_execute(&mut self, asset_ids: &[U256]) -> bool {
+        let key = Self::canonical_key(asset_ids);
+        let now = Instant::now();
+
+        if let Some(last_exec) = self.recent_signals.get(&key)
+            && now.duration_since(*last_exec) < self.cooldown
+        {
+            return false; // Still in cooldown
+        }
+
+        self.recent_signals.insert(key, now);
+        true
+    }
+
+    /// Periodic garbage collection: remove entries older than 2x cooldown.
+    pub fn gc(&mut self) {
+        let cutoff = self.cooldown * 2;
+        let now = Instant::now();
+        self.recent_signals
+            .retain(|_, last| now.duration_since(*last) < cutoff);
+    }
+
+    /// Create a canonical hash key from all asset IDs so any permutation maps to the same entry.
+    /// This supports multi-leg strategies (e.g. 3+ leg basket shorts).
+    fn canonical_key(asset_ids: &[U256]) -> u64 {
+        let mut sorted_ids = asset_ids.to_vec();
+        sorted_ids.sort_unstable();
+
+        let mut hasher = DefaultHasher::new();
+        for id in sorted_ids {
+            // Hash the string representation or underlying bytes.
+            // Converting to string is an easy deterministic way to hash the U256.
+            id.to_string().hash(&mut hasher);
+        }
+        hasher.finish()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_dedup_blocks_duplicate_pair() {
+        let mut dedup = SignalDeduplicator::new(30);
+        let ids = vec![U256::from(1), U256::from(2)];
+
+        assert!(dedup.try_execute(&ids)); // First attempt: allowed
+        assert!(!dedup.try_execute(&ids)); // Second attempt: blocked
+    }
+
+    #[test]
+    fn test_dedup_canonical_order() {
+        let mut dedup = SignalDeduplicator::new(30);
+        let ids_ab = vec![U256::from(1), U256::from(2)];
+        let ids_ba = vec![U256::from(2), U256::from(1)];
+
+        assert!(dedup.try_execute(&ids_ab)); // allowed
+        assert!(!dedup.try_execute(&ids_ba)); // blocked — same canonical pair
+    }
+
+    #[test]
+    fn test_dedup_multi_leg() {
+        let mut dedup = SignalDeduplicator::new(30);
+        let ids_1 = vec![U256::from(1), U256::from(2), U256::from(3)];
+        let ids_2 = vec![U256::from(3), U256::from(2), U256::from(1)];
+
+        assert!(dedup.try_execute(&ids_1)); // allowed
+        assert!(!dedup.try_execute(&ids_2)); // blocked — same basket
+    }
+
+    #[test]
+    fn test_dedup_allows_after_cooldown() {
+        let mut dedup = SignalDeduplicator::new(0); // Zero-second cooldown
+        let ids = vec![U256::from(1), U256::from(2)];
+
+        assert!(dedup.try_execute(&ids));
+        // With 0s cooldown, the next call should pass immediately
+        std::thread::sleep(std::time::Duration::from_millis(10));
+        assert!(dedup.try_execute(&ids));
+    }
+
+    #[test]
+    fn test_dedup_different_pairs_allowed() {
+        let mut dedup = SignalDeduplicator::new(30);
+        let pair_a = vec![U256::from(1), U256::from(2)];
+        let pair_b = vec![U256::from(3), U256::from(4)];
+
+        assert!(dedup.try_execute(&pair_a));
+        assert!(dedup.try_execute(&pair_b)); // Different pair: allowed
+    }
+
+    #[test]
+    fn test_gc_removes_old_entries() {
+        let mut dedup = SignalDeduplicator::new(0); // 0s cooldown → 0s gc cutoff
+        let ids = vec![U256::from(1), U256::from(2)];
+
+        dedup.try_execute(&ids);
+        std::thread::sleep(std::time::Duration::from_millis(10));
+        dedup.gc();
+
+        assert!(dedup.recent_signals.is_empty());
+    }
+}