hyper_agent_core/

live_executor.rs

use std::collections::HashMap;
use std::sync::Arc;

use async_trait::async_trait;

use crate::config::AppConfig;
use crate::credentials::CredentialResolver;
use crate::executor::{ExecutorError, OrderParams, OrderResult, OrderSubmitter};
use hyper_exchange::{sign_l1_action, ExchangeClient, Signer};
use hyper_keyring::simple::SimpleKeyring;
use hyper_keyring::Keyring;

/// Cache of asset name → index mappings from the Hyperliquid exchange meta endpoint.
///
/// Used by [`LiveExecutor`] to resolve human-readable market symbols (e.g. "BTC-PERP")
/// into the numeric asset indices required by the L1 order action.
#[derive(Clone)]
pub struct AssetMetaCache {
    coin_to_index: HashMap<String, u32>,
}

impl AssetMetaCache {
    /// Fetch asset metadata from the exchange info endpoint and build the cache.
    pub async fn fetch(client: &ExchangeClient) -> Result<Self, String> {
        let meta = client
            .post_info(serde_json::json!({ "type": "meta" }))
            .await
            .map_err(|e| format!("Failed to fetch exchange meta: {}", e))?;
        let universe = meta
            .get("universe")
            .and_then(|u| u.as_array())
            .ok_or_else(|| "Invalid meta response: missing universe".to_string())?;
        let mut coin_to_index = HashMap::new();
        for (idx, asset) in universe.iter().enumerate() {
            if let Some(name) = asset.get("name").and_then(|n| n.as_str()) {
                coin_to_index.insert(name.to_uppercase(), idx as u32);
            }
        }
        Ok(Self { coin_to_index })
    }

    /// Create a cache from a pre-built map (useful for testing).
    pub fn from_map(map: HashMap<String, u32>) -> Self {
        Self { coin_to_index: map }
    }

    /// Resolve a market symbol to its asset index.
    ///
    /// Strips common suffixes (-PERP, -USDC, -USD) and uppercases before lookup.
    pub fn resolve(&self, symbol: &str) -> Option<u32> {
        let coin = symbol
            .to_uppercase()
            .replace("-PERP", "")
            .replace("-USDC", "")
            .replace("-USD", "");
        self.coin_to_index.get(&coin).copied()
    }
}

/// Live order executor that submits EIP-712 signed actions to Hyperliquid L1.
pub struct LiveExecutor {
    signer: Arc<dyn Signer>,
    agent_address: String,
    vault_address: Option<String>,
    is_mainnet: bool,
    client: ExchangeClient,
    meta_cache: AssetMetaCache,
}

impl LiveExecutor {
    pub fn new(
        signer: Arc<dyn Signer>,
        agent_address: String,
        vault_address: Option<String>,
        is_mainnet: bool,
        client: ExchangeClient,
        meta_cache: AssetMetaCache,
    ) -> Self {
        Self {
            signer,
            agent_address,
            vault_address,
            is_mainnet,
            client,
            meta_cache,
        }
    }

    /// Place a trigger order (stop-loss or take-profit) on Hyperliquid.
    ///
    /// `tpsl` should be `"sl"` for stop-loss or `"tp"` for take-profit.
    /// The order fires as a market order when the trigger price is hit.
    pub async fn place_trigger_order(
        &self,
        symbol: &str,
        side: &str, // "buy" or "sell" (opposite of position side)
        size: f64,
        trigger_price: f64,
        tpsl: &str, // "sl" or "tp"
    ) -> Result<OrderResult, ExecutorError> {
        let asset_idx = self.meta_cache.resolve(symbol).ok_or_else(|| {
            ExecutorError::Execution(format!("Asset '{}' not found in exchange universe", symbol))
        })?;

        let is_buy = side == "buy";
        let nonce = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap()
            .as_millis() as u64;

        let action = serde_json::json!({
            "type": "order",
            "orders": [{
                "a": asset_idx,
                "b": is_buy,
                "p": format!("{}", trigger_price),
                "s": format!("{}", size),
                "r": true,
                "t": {
                    "trigger": {
                        "triggerPx": format!("{}", trigger_price),
                        "isMarket": true,
                        "tpsl": tpsl
                    }
                }
            }],
            "grouping": "na"
        });

        let signature = sign_l1_action(
            self.signer.as_ref(),
            &self.agent_address,
            &action,
            nonce,
            self.is_mainnet,
            self.vault_address.as_deref(),
        )
        .map_err(|e| ExecutorError::Execution(format!("Signing failed: {}", e)))?;

        let result = self
            .client
            .post_action(action, &signature, nonce, self.vault_address.as_deref())
            .await
            .map_err(|e| ExecutorError::Execution(format!("Exchange API error: {}", e)))?;

        let api_status = result
            .get("status")
            .and_then(|s| s.as_str())
            .unwrap_or("unknown");

        if api_status != "ok" {
            return Err(ExecutorError::Execution(format!(
                "Trigger order rejected: {}",
                result
            )));
        }

        // Parse fill info from exchange response
        let status_entry = result
            .get("response")
            .and_then(|r| r.get("data"))
            .and_then(|d| d.get("statuses"))
            .and_then(|s| s.as_array())
            .and_then(|a| a.first());

        let (actual_order_id, actual_fill_price, actual_fill_size) =
            if let Some(entry) = status_entry {
                if let Some(filled) = entry.get("filled") {
                    let oid = filled
                        .get("oid")
                        .and_then(|o| o.as_u64())
                        .map(|o| o.to_string());
                    let avg_px = filled
                        .get("avgPx")
                        .and_then(|p| p.as_str())
                        .and_then(|s| s.parse::<f64>().ok());
                    let total_sz = filled
                        .get("totalSz")
                        .and_then(|s| s.as_str())
                        .and_then(|s| s.parse::<f64>().ok());
                    (
                        oid.unwrap_or_else(|| uuid::Uuid::new_v4().to_string()),
                        avg_px.unwrap_or(trigger_price),
                        total_sz.unwrap_or(size),
                    )
                } else if let Some(resting) = entry.get("resting") {
                    let oid = resting
                        .get("oid")
                        .and_then(|o| o.as_u64())
                        .map(|o| o.to_string());
                    (
                        oid.unwrap_or_else(|| uuid::Uuid::new_v4().to_string()),
                        trigger_price, // Not filled yet, use trigger price
                        0.0,           // Resting order has zero fill
                    )
                } else {
                    (uuid::Uuid::new_v4().to_string(), trigger_price, size)
                }
            } else {
                (uuid::Uuid::new_v4().to_string(), trigger_price, size)
            };

        // Zero-fill (actual_fill_size == 0.0) is excluded from partial_fill because
        // trigger orders are expected to rest unfilled on the book until the trigger
        // price is hit. A zero-fill simply means the order is resting, not partially filled.
        let status = if actual_fill_size < size * 0.99 && actual_fill_size > 0.0 {
            tracing::warn!(
                order_id = %actual_order_id,
                filled = actual_fill_size,
                requested = size,
                "Partial fill detected on trigger order"
            );
            "partial_fill".to_string()
        } else if actual_fill_size == 0.0 {
            "resting".to_string()
        } else {
            format!("trigger_{}", tpsl)
        };

        Ok(OrderResult {
            order_id: actual_order_id,
            filled_price: actual_fill_price,
            filled_size: actual_fill_size,
            requested_size: size,
            status,
        })
    }

    /// Fetch the mid price for a given symbol from the L2 orderbook.
    async fn fetch_mid_price(&self, symbol: &str) -> Result<f64, String> {
        let coin = symbol
            .to_uppercase()
            .replace("-PERP", "")
            .replace("-USDC", "")
            .replace("-USD", "");

        let body = serde_json::json!({ "type": "l2Book", "coin": coin });
        let resp = self
            .client
            .post_info(body)
            .await
            .map_err(|e| format!("L2 book fetch failed: {}", e))?;

        let levels = resp
            .get("levels")
            .and_then(|l| l.as_array())
            .ok_or("Invalid L2 book response")?;

        if levels.len() < 2 {
            return Err("Incomplete L2 book".into());
        }

        let best_bid = levels[0]
            .as_array()
            .and_then(|bids| bids.first())
            .and_then(|b| b.get("px"))
            .and_then(|p| p.as_str())
            .and_then(|s| s.parse::<f64>().ok())
            .ok_or("Cannot parse best bid")?;

        let best_ask = levels[1]
            .as_array()
            .and_then(|asks| asks.first())
            .and_then(|a| a.get("px"))
            .and_then(|p| p.as_str())
            .and_then(|s| s.parse::<f64>().ok())
            .ok_or("Cannot parse best ask")?;

        Ok((best_bid + best_ask) / 2.0)
    }
}

#[async_trait]
impl OrderSubmitter for LiveExecutor {
    async fn place_order(&self, params: &OrderParams) -> Result<OrderResult, ExecutorError> {
        let asset_idx = self.meta_cache.resolve(&params.market).ok_or_else(|| {
            ExecutorError::Execution(format!(
                "Asset '{}' not found in exchange universe",
                params.market
            ))
        })?;

        let is_buy = params.side == "buy";
        let nonce = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap()
            .as_millis() as u64;

        // Determine price and order type
        let (effective_price, order_type) = if let Some(p) = params.price {
            // Limit order with explicit price -> GTC
            (p, serde_json::json!({ "limit": { "tif": "Gtc" } }))
        } else {
            // Market order -> IOC with slippage
            let mid_price = self.fetch_mid_price(&params.market).await.map_err(|e| {
                ExecutorError::Execution(format!(
                    "Failed to fetch mid price for market order: {}",
                    e
                ))
            })?;
            let slippage = 0.005; // 0.5%
            let slipped = if is_buy {
                mid_price * (1.0 + slippage)
            } else {
                mid_price * (1.0 - slippage)
            };
            (slipped, serde_json::json!({ "limit": { "tif": "Ioc" } }))
        };

        let action = serde_json::json!({
            "type": "order",
            "orders": [{
                "a": asset_idx,
                "b": is_buy,
                "p": format!("{}", effective_price),
                "s": format!("{}", params.size),
                "r": false,
                "t": order_type
            }],
            "grouping": "na"
        });

        let signature = sign_l1_action(
            self.signer.as_ref(),
            &self.agent_address,
            &action,
            nonce,
            self.is_mainnet,
            self.vault_address.as_deref(),
        )
        .map_err(|e| ExecutorError::Execution(format!("Signing failed: {}", e)))?;

        let result = self
            .client
            .post_action(action, &signature, nonce, self.vault_address.as_deref())
            .await
            .map_err(|e| ExecutorError::Execution(format!("Exchange API error: {}", e)))?;

        let api_status = result
            .get("status")
            .and_then(|s| s.as_str())
            .unwrap_or("unknown");

        if api_status != "ok" {
            return Err(ExecutorError::Execution(format!(
                "Exchange rejected order: {}",
                result
            )));
        }

        // Parse fill info from exchange response
        let status_entry = result
            .get("response")
            .and_then(|r| r.get("data"))
            .and_then(|d| d.get("statuses"))
            .and_then(|s| s.as_array())
            .and_then(|a| a.first());

        let (actual_order_id, actual_fill_price, actual_fill_size) =
            if let Some(entry) = status_entry {
                if let Some(filled) = entry.get("filled") {
                    let oid = filled
                        .get("oid")
                        .and_then(|o| o.as_u64())
                        .map(|o| o.to_string());
                    let avg_px = filled
                        .get("avgPx")
                        .and_then(|p| p.as_str())
                        .and_then(|s| s.parse::<f64>().ok());
                    let total_sz = filled
                        .get("totalSz")
                        .and_then(|s| s.as_str())
                        .and_then(|s| s.parse::<f64>().ok());
                    (
                        oid.unwrap_or_else(|| uuid::Uuid::new_v4().to_string()),
                        avg_px.unwrap_or(effective_price),
                        total_sz.unwrap_or(params.size),
                    )
                } else if let Some(resting) = entry.get("resting") {
                    let oid = resting
                        .get("oid")
                        .and_then(|o| o.as_u64())
                        .map(|o| o.to_string());
                    (
                        oid.unwrap_or_else(|| uuid::Uuid::new_v4().to_string()),
                        effective_price, // Not filled yet, use limit price
                        0.0,             // Resting order has zero fill
                    )
                } else {
                    (
                        uuid::Uuid::new_v4().to_string(),
                        effective_price,
                        params.size,
                    )
                }
            } else {
                (
                    uuid::Uuid::new_v4().to_string(),
                    effective_price,
                    params.size,
                )
            };

        let status = if actual_fill_size < params.size * 0.99 && actual_fill_size > 0.0 {
            tracing::warn!(
                order_id = %actual_order_id,
                filled = actual_fill_size,
                requested = params.size,
                "Partial fill detected"
            );
            "partial_fill".to_string()
        } else if actual_fill_size >= params.size * 0.99 {
            "filled".to_string()
        } else if actual_fill_size == 0.0 {
            "resting".to_string()
        } else {
            api_status.to_string()
        };

        Ok(OrderResult {
            order_id: actual_order_id,
            filled_price: actual_fill_price,
            filled_size: actual_fill_size,
            requested_size: params.size,
            status,
        })
    }

    async fn cancel_order(&self, order_id: &str) -> Result<(), ExecutorError> {
        // Parse order_id as u64 (Hyperliquid uses numeric OIDs)
        let oid: u64 = order_id.parse().map_err(|_| {
            ExecutorError::Execution(format!("Invalid order ID '{}': must be numeric", order_id))
        })?;

        let nonce = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap()
            .as_millis() as u64;

        // Query open orders to find the asset index for this order
        let body = serde_json::json!({
            "type": "openOrders",
            "user": self.agent_address,
        });
        let orders =
            self.client.post_info(body).await.map_err(|e| {
                ExecutorError::Execution(format!("Failed to fetch open orders: {}", e))
            })?;

        // Find the order to get its coin/asset index
        let order = orders
            .as_array()
            .and_then(|arr| {
                arr.iter()
                    .find(|o| o.get("oid").and_then(|id| id.as_u64()) == Some(oid))
            })
            .ok_or_else(|| {
                ExecutorError::Execution(format!("Order {} not found in open orders", order_id))
            })?;

        let coin = order.get("coin").and_then(|c| c.as_str()).unwrap_or("");
        let asset_idx = self
            .meta_cache
            .resolve(&format!("{}-PERP", coin))
            .ok_or_else(|| {
                ExecutorError::Execution(format!("Asset '{}' not in meta cache", coin))
            })?;

        let action = serde_json::json!({
            "type": "cancel",
            "cancels": [{ "a": asset_idx, "o": oid }]
        });

        let signature = sign_l1_action(
            self.signer.as_ref(),
            &self.agent_address,
            &action,
            nonce,
            self.is_mainnet,
            self.vault_address.as_deref(),
        )
        .map_err(|e| ExecutorError::Execution(format!("Signing failed: {}", e)))?;

        let result = self
            .client
            .post_action(action, &signature, nonce, self.vault_address.as_deref())
            .await
            .map_err(|e| ExecutorError::Execution(format!("Cancel API error: {}", e)))?;

        let status = result
            .get("status")
            .and_then(|s| s.as_str())
            .unwrap_or("unknown");
        if status != "ok" {
            return Err(ExecutorError::Execution(format!(
                "Cancel rejected: {}",
                result
            )));
        }

        tracing::info!(order_id = %order_id, "[live] order cancelled");
        Ok(())
    }
}

/// Build a [`LiveExecutor`] from the app configuration.
///
/// Resolves the Hyperliquid private key from environment variables or TOML
/// config, derives the agent address, constructs a signer via
/// [`SimpleKeyring`], fetches the exchange asset metadata, and returns a
/// ready-to-use [`LiveExecutor`] (which also implements [`OrderSubmitter`]).
pub async fn build_live_executor(
    config: &AppConfig,
) -> Result<Arc<LiveExecutor>, Box<dyn std::error::Error>> {
    let resolver = CredentialResolver::new(config.credentials.clone());

    // 1. Get private key
    let private_key = resolver.hyperliquid_key().ok_or(
        "Live mode requires HYPERLIQUID_PRIVATE_KEY env var or credentials.hyperliquid_private_key in config.toml",
    )?;

    // 2. Get address from key
    let address = resolver
        .hyperliquid_address()
        .ok_or("No Hyperliquid private key configured")?
        .map_err(|e| format!("Address derivation failed: {}", e))?;

    // 3. Build signer (SimpleKeyring implements hyper_exchange::Signer)
    let mut keyring = SimpleKeyring::new();
    keyring
        .add_accounts(&[private_key])
        .map_err(|e| format!("Failed to import key: {}", e))?;
    let signer: Arc<dyn hyper_exchange::Signer> = Arc::new(keyring);

    // 4. Build exchange client
    let client = ExchangeClient::new(config.exchange.is_mainnet);

    // 5. Fetch asset meta cache
    let meta_cache = AssetMetaCache::fetch(&client)
        .await
        .map_err(|e| format!("Failed to fetch exchange meta: {}", e))?;

    // 6. Construct LiveExecutor
    Ok(Arc::new(LiveExecutor::new(
        signer,
        address,
        config.exchange.vault_address.clone(),
        config.exchange.is_mainnet,
        client,
        meta_cache,
    )))
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn meta_cache_resolve_btc_perp() {
        let mut map = HashMap::new();
        map.insert("BTC".to_string(), 0);
        map.insert("ETH".to_string(), 1);
        let cache = AssetMetaCache::from_map(map);
        assert_eq!(cache.resolve("BTC-PERP"), Some(0));
        assert_eq!(cache.resolve("ETH-PERP"), Some(1));
        assert_eq!(cache.resolve("SOL-PERP"), None);
    }

    #[test]
    fn meta_cache_resolve_strips_suffixes() {
        let mut map = HashMap::new();
        map.insert("BTC".to_string(), 0);
        let cache = AssetMetaCache::from_map(map);
        assert_eq!(cache.resolve("BTC-USD"), Some(0));
        assert_eq!(cache.resolve("BTC-USDC"), Some(0));
        assert_eq!(cache.resolve("BTC-PERP"), Some(0));
    }

    #[test]
    fn meta_cache_resolve_case_insensitive() {
        let mut map = HashMap::new();
        map.insert("BTC".to_string(), 0);
        let cache = AssetMetaCache::from_map(map);
        assert_eq!(cache.resolve("btc-perp"), Some(0));
    }

    #[test]
    fn meta_cache_resolve_for_cancel() {
        let mut map = HashMap::new();
        map.insert("BTC".to_string(), 0);
        let cache = AssetMetaCache::from_map(map);
        assert_eq!(cache.resolve("BTC-PERP"), Some(0));
    }

    #[test]
    fn invalid_order_id_is_error() {
        let result: Result<u64, _> = "not-a-number".parse();
        assert!(result.is_err());
    }

    #[test]
    fn market_order_slippage_buy() {
        let mid = 50000.0_f64;
        let slippage = 0.005;
        let slipped = mid * (1.0 + slippage);
        assert!((slipped - 50250.0).abs() < 0.01);
    }

    #[test]
    fn market_order_slippage_sell() {
        let mid = 50000.0_f64;
        let slippage = 0.005;
        let slipped = mid * (1.0 - slippage);
        assert!((slipped - 49750.0).abs() < 0.01);
    }

    #[test]
    fn parse_filled_response_extracts_real_price_and_size() {
        let response: serde_json::Value = serde_json::json!({
            "status": "ok",
            "response": {
                "type": "order",
                "data": {
                    "statuses": [{
                        "filled": {
                            "totalSz": "0.01",
                            "avgPx": "65123.5",
                            "oid": 12345
                        }
                    }]
                }
            }
        });

        let status_entry = response
            .get("response")
            .and_then(|r| r.get("data"))
            .and_then(|d| d.get("statuses"))
            .and_then(|s| s.as_array())
            .and_then(|a| a.first());

        let entry = status_entry.unwrap();
        let filled = entry.get("filled").unwrap();
        let avg_px: f64 = filled
            .get("avgPx")
            .unwrap()
            .as_str()
            .unwrap()
            .parse()
            .unwrap();
        let total_sz: f64 = filled
            .get("totalSz")
            .unwrap()
            .as_str()
            .unwrap()
            .parse()
            .unwrap();
        let oid = filled.get("oid").unwrap().as_u64().unwrap();

        assert!((avg_px - 65123.5).abs() < 0.01);
        assert!((total_sz - 0.01).abs() < 0.0001);
        assert_eq!(oid, 12345);
    }

    #[test]
    fn parse_resting_response_uses_fallback_price() {
        let response: serde_json::Value = serde_json::json!({
            "status": "ok",
            "response": {
                "type": "order",
                "data": {
                    "statuses": [{
                        "resting": {
                            "oid": 67890
                        }
                    }]
                }
            }
        });

        let status_entry = response
            .get("response")
            .and_then(|r| r.get("data"))
            .and_then(|d| d.get("statuses"))
            .and_then(|s| s.as_array())
            .and_then(|a| a.first());

        let entry = status_entry.unwrap();
        assert!(entry.get("filled").is_none());
        let resting = entry.get("resting").unwrap();
        let oid = resting.get("oid").unwrap().as_u64().unwrap();
        assert_eq!(oid, 67890);
        // avgPx not available for resting orders -> should fall back to effective_price
        assert!(resting.get("avgPx").is_none());
    }

    #[test]
    fn parse_empty_statuses_uses_fallback() {
        let response: serde_json::Value = serde_json::json!({
            "status": "ok",
            "response": {
                "type": "order",
                "data": {
                    "statuses": []
                }
            }
        });

        let status_entry = response
            .get("response")
            .and_then(|r| r.get("data"))
            .and_then(|d| d.get("statuses"))
            .and_then(|s| s.as_array())
            .and_then(|a| a.first());

        assert!(status_entry.is_none());
    }
}