hypercall/hypercore/

hydromancer_feed.rs

//! Hydromancer WebSocket feed for real-time order updates, fills, and positions.
//!
//! Subscribes to `userOrderUpdates` and `userFills` for tracked accounts via
//! Hydromancer's WebSocket API. Maintains current position state per account.
//!
//! On boot and reconnect: fetches `portfolioState` from Hydromancer API to seed
//! position snapshots and catch missed fills. Uses the portfolio endpoint (not raw
//! `clearinghouseState`) so that spot USDC balances are included in account equity
//! for unified accounts.

use std::collections::HashMap;
use std::sync::Arc;
use std::time::{Duration, Instant};

use futures::{SinkExt, StreamExt};
use serde::{Deserialize, Serialize};
use tokio::sync::{broadcast, RwLock};
use tokio_tungstenite::tungstenite::Message;
use tracing::{debug, error, info, warn};

use hypercall_types::WalletAddress;

// ---------------------------------------------------------------------------
// Types
// ---------------------------------------------------------------------------

/// A perp position snapshot for one coin.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PerpPosition {
    pub coin: String,
    pub size: f64,
    pub entry_price: Option<f64>,
    pub unrealized_pnl: f64,
    pub margin_used: f64,
}

/// Full account state: equity + positions.
#[derive(Debug, Clone, Default)]
pub struct AccountState {
    pub account_value: f64,
    pub total_margin_used: f64,
    pub withdrawable: f64,
    pub positions: Vec<PerpPosition>,
    pub last_updated_ms: u64,
}

/// An order update event from Hydromancer WS.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct OrderUpdate {
    pub coin: String,
    pub oid: Option<u64>,
    pub side: Option<String>,
    pub sz: Option<String>,
    pub limit_px: Option<String>,
    pub order_type: Option<String>,
    pub status: Option<String>,
    pub filled_sz: Option<String>,
    pub avg_px: Option<String>,
    pub timestamp: Option<u64>,
}

/// A fill event from Hydromancer WS.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct FillEvent {
    pub coin: String,
    pub side: String,
    pub px: String,
    pub sz: String,
    pub time: u64,
    pub oid: Option<u64>,
    pub fee: Option<String>,
}

/// Events emitted by the feed service.
#[derive(Debug, Clone)]
pub enum HydromancerEvent {
    PositionSnapshot {
        account: WalletAddress,
        state: AccountState,
    },
    OrderUpdate {
        account: WalletAddress,
        update: OrderUpdate,
    },
    Fill {
        account: WalletAddress,
        fill: FillEvent,
    },
}

// ---------------------------------------------------------------------------
// Service
// ---------------------------------------------------------------------------

pub struct HydromancerFeedService {
    ws_url: String,
    accounts: Arc<RwLock<Vec<WalletAddress>>>,
    positions: Arc<RwLock<HashMap<WalletAddress, AccountState>>>,
    event_tx: broadcast::Sender<HydromancerEvent>,
    hl_info_url: String,
    /// Hydromancer API URL for portfolioState queries (includes spot balances).
    hydromancer_api_url: String,
    /// Bearer token for Hydromancer API authentication.
    hydromancer_api_key: Option<String>,
}

#[async_trait::async_trait]
impl hypercall_api::state::DirectiveHydromancerFeed for HydromancerFeedService {
    async fn add_account(&self, account: WalletAddress) {
        self.add_account(account).await;
    }

    async fn wait_for_fills(
        &self,
        account: &WalletAddress,
        timeout: Duration,
    ) -> Vec<hypercall_api::state::DirectiveHydromancerFill> {
        self.wait_for_fills(account, timeout)
            .await
            .into_iter()
            .map(|fill| hypercall_api::state::DirectiveHydromancerFill {
                coin: fill.coin,
                side: fill.side,
                px: fill.px,
                sz: fill.sz,
                time: fill.time,
            })
            .collect()
    }
}

impl HydromancerFeedService {
    pub fn new(
        ws_url: String,
        hl_info_url: String,
        hydromancer_api_url: String,
        hydromancer_api_key: Option<String>,
        initial_accounts: Vec<WalletAddress>,
    ) -> Self {
        let (event_tx, _) = broadcast::channel(1024);
        Self {
            ws_url,
            accounts: Arc::new(RwLock::new(initial_accounts)),
            positions: Arc::new(RwLock::new(HashMap::new())),
            event_tx,
            hl_info_url,
            hydromancer_api_url,
            hydromancer_api_key,
        }
    }

    pub fn subscribe(&self) -> broadcast::Receiver<HydromancerEvent> {
        self.event_tx.subscribe()
    }

    pub async fn add_account(&self, account: WalletAddress) {
        let mut accounts = self.accounts.write().await;
        if !accounts.contains(&account) {
            accounts.push(account);
            info!(account = %account, "HydromancerFeed: added account");
        }
    }

    /// Get current position state for an account. Returns None if not yet loaded.
    pub async fn get_positions(&self, account: &WalletAddress) -> Option<AccountState> {
        self.positions.read().await.get(account).cloned()
    }

    /// Get current position for a specific coin. Returns 0.0 if no position.
    pub async fn get_position_size(&self, account: &WalletAddress, coin: &str) -> f64 {
        self.positions
            .read()
            .await
            .get(account)
            .and_then(|state| state.positions.iter().find(|p| p.coin == coin))
            .map(|p| p.size)
            .unwrap_or(0.0)
    }

    /// Wait for a fill event for a specific account, with timeout.
    pub async fn wait_for_fills(
        &self,
        account: &WalletAddress,
        timeout: Duration,
    ) -> Vec<FillEvent> {
        let mut rx = self.event_tx.subscribe();
        let deadline = Instant::now() + timeout;
        let mut fills = Vec::new();

        loop {
            let remaining = deadline.saturating_duration_since(Instant::now());
            if remaining.is_zero() {
                break;
            }

            match tokio::time::timeout(remaining, rx.recv()).await {
                Ok(Ok(HydromancerEvent::Fill { account: a, fill })) if a == *account => {
                    fills.push(fill);
                    tokio::time::sleep(Duration::from_millis(200)).await;
                }
                Ok(Ok(_)) => continue,
                Ok(Err(_)) => break,
                Err(_) => break,
            }
        }

        fills
    }

    // -----------------------------------------------------------------------
    // Lifecycle
    // -----------------------------------------------------------------------

    /// Start the feed service. Runs until shutdown.
    pub async fn start(&self, mut shutdown_rx: broadcast::Receiver<()>) {
        const BASE_BACKOFF: Duration = Duration::from_secs(1);
        const MAX_BACKOFF: Duration = Duration::from_secs(30);
        let mut consecutive_failures: u32 = 0;

        loop {
            let before = Instant::now();

            // On every (re)connect: fetch full position snapshots + recent fills
            self.reconcile_on_connect().await;

            tokio::select! {
                _ = shutdown_rx.recv() => {
                    info!("HydromancerFeed: shutdown signal received");
                    return;
                }
                _ = self.run_connection() => {
                    if before.elapsed() > Duration::from_secs(60) {
                        consecutive_failures = 0;
                    }
                    consecutive_failures += 1;
                    let backoff = std::cmp::min(
                        BASE_BACKOFF * 2u32.saturating_pow(consecutive_failures - 1),
                        MAX_BACKOFF,
                    );
                    warn!(
                        failures = consecutive_failures,
                        backoff_ms = backoff.as_millis() as u64,
                        "HydromancerFeed: connection lost, reconnecting"
                    );
                    tokio::select! {
                        _ = shutdown_rx.recv() => {
                            info!("HydromancerFeed: shutdown during backoff");
                            return;
                        }
                        _ = tokio::time::sleep(backoff) => {}
                    }
                }
            }
        }
    }

    // -----------------------------------------------------------------------
    // REST reconciliation (boot + reconnect)
    // -----------------------------------------------------------------------

    async fn reconcile_on_connect(&self) {
        let accts = self.accounts.read().await.clone();
        if accts.is_empty() {
            return;
        }

        let client = match reqwest::Client::builder()
            .timeout(Duration::from_secs(10))
            .build()
        {
            Ok(c) => c,
            Err(e) => {
                error!("HydromancerFeed: failed to create HTTP client: {}", e);
                return;
            }
        };

        let now_ms = crate::shared::order_types::get_timestamp_millis();

        for account in &accts {
            // 1. Fetch portfolioState (perp + spot) via Hydromancer API
            let body = serde_json::json!({
                "type": "portfolioState",
                "user": format!("{}", account),
            });

            let mut request = client.post(&self.hydromancer_api_url).json(&body);
            if let Some(ref key) = self.hydromancer_api_key {
                request = request.header("Authorization", format!("Bearer {}", key));
            }

            match request.send().await {
                Ok(resp) => {
                    if let Ok(state_json) = resp.json::<serde_json::Value>().await {
                        let account_state = parse_portfolio_state(&state_json, now_ms);
                        self.positions
                            .write()
                            .await
                            .insert(*account, account_state.clone());
                        let _ = self.event_tx.send(HydromancerEvent::PositionSnapshot {
                            account: *account,
                            state: account_state,
                        });
                    }
                }
                Err(e) => {
                    warn!(account = %account, "HydromancerFeed: portfolioState fetch failed: {}", e);
                }
            }

            // 2. Fetch recent fills (last 60s)
            let fills_body = serde_json::json!({
                "type": "userFillsByTime",
                "user": format!("{}", account),
                "startTime": now_ms.saturating_sub(60_000),
            });

            if let Ok(resp) = client
                .post(&self.hl_info_url)
                .json(&fills_body)
                .send()
                .await
            {
                if let Ok(fills_json) = resp.json::<Vec<serde_json::Value>>().await {
                    for fill_val in fills_json {
                        if let Ok(fill) = serde_json::from_value::<FillEvent>(fill_val) {
                            let _ = self.event_tx.send(HydromancerEvent::Fill {
                                account: *account,
                                fill,
                            });
                        }
                    }
                }
            }
        }

        info!(
            accounts = accts.len(),
            "HydromancerFeed: reconciliation complete"
        );
    }

    // -----------------------------------------------------------------------
    // WebSocket connection
    // -----------------------------------------------------------------------

    async fn run_connection(&self) {
        let redacted_url = self.ws_url.split("?token=").next().unwrap_or(&self.ws_url);
        info!(url = %redacted_url, "HydromancerFeed: connecting");

        let (ws_stream, _) = match tokio_tungstenite::connect_async(&self.ws_url).await {
            Ok(conn) => {
                info!("HydromancerFeed: connected");
                conn
            }
            Err(e) => {
                error!("HydromancerFeed: connection failed: {}", e);
                return;
            }
        };

        let (mut write, mut read) = ws_stream.split();

        // Read the "connected" message
        match tokio::time::timeout(Duration::from_secs(5), read.next()).await {
            Ok(Some(Ok(Message::Text(text)))) => {
                debug!("HydromancerFeed: {}", text);
            }
            _ => {
                error!("HydromancerFeed: no connected message received");
                return;
            }
        }

        // Subscribe for all tracked accounts
        let accts = self.accounts.read().await;
        let addresses: Vec<String> = accts.iter().map(|a| format!("{}", a)).collect();
        drop(accts);

        if addresses.is_empty() {
            warn!("HydromancerFeed: no accounts to subscribe to");
        }

        for sub_type in &["userOrderUpdates", "userFills"] {
            let sub = serde_json::json!({
                "method": "subscribe",
                "subscription": {
                    "type": sub_type,
                    "addresses": addresses,
                }
            });
            let msg = serde_json::to_string(&sub).expect("serialize subscription");
            if let Err(e) = write.send(Message::Text(msg)).await {
                error!(
                    sub_type = sub_type,
                    "HydromancerFeed: subscribe failed: {}", e
                );
                return;
            }
        }

        // Read subscription confirmations
        for _ in 0..2 {
            match tokio::time::timeout(Duration::from_secs(5), read.next()).await {
                Ok(Some(Ok(Message::Text(text)))) => {
                    debug!("HydromancerFeed: sub response: {}", text);
                }
                _ => {
                    warn!("HydromancerFeed: subscription confirmation timeout");
                }
            }
        }

        info!(
            accounts = addresses.len(),
            "HydromancerFeed: subscribed to userOrderUpdates + userFills"
        );

        // Event loop
        const WS_READ_TIMEOUT: Duration = Duration::from_secs(60);

        loop {
            match tokio::time::timeout(WS_READ_TIMEOUT, read.next()).await {
                Ok(Some(Ok(Message::Text(text)))) => {
                    self.handle_message(&text);
                }
                Ok(Some(Ok(Message::Ping(data)))) => {
                    if let Err(e) = write.send(Message::Pong(data)).await {
                        error!("HydromancerFeed: pong failed: {}", e);
                        return;
                    }
                }
                Ok(Some(Ok(Message::Close(_)))) => {
                    info!("HydromancerFeed: server closed connection");
                    return;
                }
                Ok(Some(Err(e))) => {
                    error!("HydromancerFeed: read error: {}", e);
                    return;
                }
                Ok(None) => {
                    info!("HydromancerFeed: stream ended");
                    return;
                }
                Err(_) => {
                    warn!(
                        "HydromancerFeed: read timeout ({}s), reconnecting",
                        WS_READ_TIMEOUT.as_secs()
                    );
                    return;
                }
                _ => {}
            }
        }
    }

    fn handle_message(&self, text: &str) {
        let msg: serde_json::Value = match serde_json::from_str(text) {
            Ok(v) => v,
            Err(e) => {
                warn!("HydromancerFeed: invalid JSON: {}", e);
                return;
            }
        };

        let channel = msg.get("channel").and_then(|v| v.as_str()).unwrap_or("");
        let data = match msg.get("data") {
            Some(d) => d,
            None => return,
        };

        if channel.starts_with("userFills") {
            self.handle_fills(data);
        } else if channel.starts_with("userOrderUpdates") {
            self.handle_order_updates(data);
        }
    }

    fn handle_fills(&self, data: &serde_json::Value) {
        let fills = match data.as_array() {
            Some(a) => a,
            None => return,
        };

        for fill_val in fills {
            let account_str = fill_val.get("user").and_then(|v| v.as_str()).unwrap_or("");
            let account = match account_str.parse::<WalletAddress>() {
                Ok(a) => a,
                Err(_) => continue,
            };
            let fill: FillEvent = match serde_json::from_value(fill_val.clone()) {
                Ok(f) => f,
                Err(_) => continue,
            };

            // Update position based on fill
            self.apply_fill_to_positions(&account, &fill);

            let _ = self.event_tx.send(HydromancerEvent::Fill { account, fill });
        }
    }

    fn handle_order_updates(&self, data: &serde_json::Value) {
        let updates = match data.as_array() {
            Some(a) => a,
            None => return,
        };

        for update_val in updates {
            let account_str = update_val
                .get("user")
                .and_then(|v| v.as_str())
                .unwrap_or("");
            let account = match account_str.parse::<WalletAddress>() {
                Ok(a) => a,
                Err(_) => continue,
            };
            let update: OrderUpdate = match serde_json::from_value(update_val.clone()) {
                Ok(u) => u,
                Err(_) => continue,
            };
            let _ = self
                .event_tx
                .send(HydromancerEvent::OrderUpdate { account, update });
        }
    }

    /// Apply a fill to the in-memory position state.
    /// This is a best-effort delta update — the full snapshot from REST is the source of truth.
    fn apply_fill_to_positions(&self, account: &WalletAddress, fill: &FillEvent) {
        let size: f64 = fill.sz.parse().unwrap_or(0.0);
        let signed_size = if fill.side == "B" { size } else { -size };

        // Try to update synchronously via try_write to avoid blocking the WS read loop.
        // If the lock is held (reconciliation running), skip — the next reconciliation
        // will correct the state anyway.
        if let Ok(mut positions) = self.positions.try_write() {
            let state = positions.entry(*account).or_default();
            if let Some(pos) = state.positions.iter_mut().find(|p| p.coin == fill.coin) {
                pos.size += signed_size;
            } else {
                state.positions.push(PerpPosition {
                    coin: fill.coin.clone(),
                    size: signed_size,
                    entry_price: fill.px.parse().ok(),
                    unrealized_pnl: 0.0,
                    margin_used: 0.0,
                });
            }
            state.last_updated_ms = fill.time;
        }
    }
}

// ---------------------------------------------------------------------------
// Helpers
// ---------------------------------------------------------------------------

/// Parse `portfolioState` response which wraps data under `clearinghouseState` and
/// `spotClearinghouseState` keys. Computes unified account value as
/// `perp_account_value + spot_usdc_balance`.
fn parse_portfolio_state(json: &serde_json::Value, now_ms: u64) -> AccountState {
    // The portfolioState response nests perp data under `clearinghouseState`.
    // Fall back to top-level keys for backward compatibility with raw clearinghouseState.
    let ch = json.get("clearinghouseState").unwrap_or(json);

    let ms = ch.get("marginSummary").unwrap_or(ch);

    let perp_account_value = ms
        .get("accountValue")
        .and_then(|v| v.as_str())
        .and_then(|s| s.parse::<f64>().ok())
        .unwrap_or(0.0);

    let total_margin_used = ms
        .get("totalMarginUsed")
        .and_then(|v| v.as_str())
        .and_then(|s| s.parse::<f64>().ok())
        .unwrap_or(0.0);

    let withdrawable = ch
        .get("withdrawable")
        .and_then(|v| v.as_str())
        .and_then(|s| s.parse::<f64>().ok())
        .unwrap_or(0.0);

    let positions = ch
        .get("assetPositions")
        .and_then(|v| v.as_array())
        .map(|arr| {
            arr.iter()
                .filter_map(|ap| {
                    let pos = ap.get("position")?;
                    let coin = pos.get("coin")?.as_str()?.to_string();
                    let size = pos.get("szi")?.as_str()?.parse::<f64>().ok()?;
                    Some(PerpPosition {
                        coin,
                        size,
                        entry_price: pos
                            .get("entryPx")
                            .and_then(|v| v.as_str())
                            .and_then(|s| s.parse().ok()),
                        unrealized_pnl: pos
                            .get("unrealizedPnl")
                            .and_then(|v| v.as_str())
                            .and_then(|s| s.parse().ok())
                            .unwrap_or(0.0),
                        margin_used: pos
                            .get("marginUsed")
                            .and_then(|v| v.as_str())
                            .and_then(|s| s.parse().ok())
                            .unwrap_or(0.0),
                    })
                })
                .collect()
        })
        .unwrap_or_default();

    // Extract spot USDC balance from spotClearinghouseState.balances
    let spot_usdc = json
        .get("spotClearinghouseState")
        .and_then(|spot| spot.get("balances"))
        .and_then(|b| b.as_array())
        .and_then(|balances| {
            balances.iter().find_map(|entry| {
                let coin = entry.get("coin")?.as_str()?;
                if coin == "USDC" {
                    entry
                        .get("total")
                        .and_then(|v| v.as_str())
                        .and_then(|s| s.parse::<f64>().ok())
                } else {
                    None
                }
            })
        })
        .unwrap_or(0.0);

    let account_value = perp_account_value + spot_usdc;

    AccountState {
        account_value,
        total_margin_used,
        withdrawable,
        positions,
        last_updated_ms: now_ms,
    }
}

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

    fn test_account() -> WalletAddress {
        WalletAddress::from_str("0x72471e6cb5be1b2db2eb00c38b7bca66c8c310c5").unwrap()
    }

    #[test]
    fn test_parse_portfolio_state() {
        let json: serde_json::Value = serde_json::from_str(
            r#"{
                "clearinghouseState": {
                    "marginSummary": {
                        "accountValue": "350.0",
                        "totalNtlPos": "3100.0",
                        "totalRawUsd": "350.0",
                        "totalMarginUsed": "77.5"
                    },
                    "withdrawable": "272.5",
                    "assetPositions": [
                        {
                            "position": {
                                "coin": "BTC",
                                "szi": "-0.04",
                                "entryPx": "77610.5",
                                "positionValue": "3100.0",
                                "unrealizedPnl": "-20.0",
                                "marginUsed": "77.5",
                                "returnOnEquity": "-0.05",
                                "liquidationPx": null,
                                "leverage": {"type": "cross", "value": 40}
                            },
                            "type": "oneWay"
                        }
                    ],
                    "time": 1776489498654
                },
                "spotClearinghouseState": {
                    "balances": [
                        {"coin": "USDC", "total": "500.0", "hold": "0.0"},
                        {"coin": "HYPE", "total": "10.0", "hold": "0.0"}
                    ]
                }
            }"#,
        )
        .unwrap();

        let state = parse_portfolio_state(&json, 1000);
        // account_value = perp(350.0) + spot_usdc(500.0) = 850.0
        assert!((state.account_value - 850.0).abs() < 0.01);
        assert!((state.total_margin_used - 77.5).abs() < 0.01);
        assert!((state.withdrawable - 272.5).abs() < 0.01);
        assert_eq!(state.positions.len(), 1);
        assert_eq!(state.positions[0].coin, "BTC");
        assert!((state.positions[0].size - (-0.04)).abs() < 0.0001);
        assert_eq!(state.positions[0].entry_price, Some(77610.5));
    }

    #[test]
    fn test_parse_portfolio_state_no_spot() {
        // portfolioState response with no spotClearinghouseState -- spot USDC defaults to 0
        let json: serde_json::Value = serde_json::from_str(
            r#"{
                "clearinghouseState": {
                    "marginSummary": {"accountValue": "100.0", "totalNtlPos": "0.0", "totalRawUsd": "100.0", "totalMarginUsed": "0.0"},
                    "withdrawable": "100.0",
                    "assetPositions": []
                }
            }"#,
        )
        .unwrap();

        let state = parse_portfolio_state(&json, 0);
        assert!((state.account_value - 100.0).abs() < 0.01);
        assert!(state.positions.is_empty());
    }

    #[test]
    fn test_parse_portfolio_state_empty() {
        let json: serde_json::Value = serde_json::from_str(
            r#"{
                "clearinghouseState": {
                    "marginSummary": {"accountValue": "0.0", "totalNtlPos": "0.0", "totalRawUsd": "0.0", "totalMarginUsed": "0.0"},
                    "withdrawable": "0.0",
                    "assetPositions": []
                },
                "spotClearinghouseState": {
                    "balances": []
                }
            }"#,
        )
        .unwrap();

        let state = parse_portfolio_state(&json, 0);
        assert_eq!(state.account_value, 0.0);
        assert!(state.positions.is_empty());
    }

    #[test]
    fn test_parse_portfolio_state_backward_compat() {
        // Backward compat: raw clearinghouseState format (no wrapper) still works
        let json: serde_json::Value = serde_json::from_str(
            r#"{
                "marginSummary": {"accountValue": "200.0", "totalNtlPos": "0.0", "totalRawUsd": "200.0", "totalMarginUsed": "50.0"},
                "withdrawable": "150.0",
                "assetPositions": []
            }"#,
        )
        .unwrap();

        let state = parse_portfolio_state(&json, 0);
        assert!((state.account_value - 200.0).abs() < 0.01);
        assert!((state.total_margin_used - 50.0).abs() < 0.01);
    }

    #[test]
    fn test_fill_event_serde() {
        let json = r#"{"coin":"BTC","side":"B","px":"78734.0","sz":"0.01","time":1776489498654}"#;
        let fill: FillEvent = serde_json::from_str(json).unwrap();
        assert_eq!(fill.coin, "BTC");
        assert_eq!(fill.side, "B");
        assert_eq!(fill.sz, "0.01");
        assert_eq!(fill.time, 1776489498654);
    }

    #[test]
    fn test_order_update_serde() {
        let json = r#"{"coin":"BTC","oid":12345,"side":"B","sz":"0.01","limitPx":"80000","status":"filled"}"#;
        let update: OrderUpdate = serde_json::from_str(json).unwrap();
        assert_eq!(update.coin, "BTC");
        assert_eq!(update.oid, Some(12345));
        assert_eq!(update.status.as_deref(), Some("filled"));
    }

    #[tokio::test]
    async fn test_service_add_account() {
        let svc = HydromancerFeedService::new(
            "ws://fake".into(),
            "http://fake".into(),
            "http://fake-hydromancer".into(),
            None,
            vec![],
        );
        let account = test_account();

        svc.add_account(account).await;
        svc.add_account(account).await; // duplicate, should be no-op

        let accounts = svc.accounts.read().await;
        assert_eq!(accounts.len(), 1);
    }

    #[tokio::test]
    async fn test_service_get_positions_empty() {
        let svc = HydromancerFeedService::new(
            "ws://fake".into(),
            "http://fake".into(),
            "http://fake-hydromancer".into(),
            None,
            vec![],
        );
        let account = test_account();
        assert!(svc.get_positions(&account).await.is_none());
        assert_eq!(svc.get_position_size(&account, "BTC").await, 0.0);
    }

    #[tokio::test]
    async fn test_service_apply_fill_to_positions() {
        let svc = HydromancerFeedService::new(
            "ws://fake".into(),
            "http://fake".into(),
            "http://fake-hydromancer".into(),
            None,
            vec![],
        );
        let account = test_account();

        // Seed empty state
        svc.positions
            .write()
            .await
            .insert(account, AccountState::default());

        // Apply a buy fill
        let fill = FillEvent {
            coin: "BTC".into(),
            side: "B".into(),
            px: "77000".into(),
            sz: "0.01".into(),
            time: 1000,
            oid: None,
            fee: None,
        };
        svc.apply_fill_to_positions(&account, &fill);

        assert!((svc.get_position_size(&account, "BTC").await - 0.01).abs() < 0.0001);

        // Apply a sell fill
        let fill2 = FillEvent {
            coin: "BTC".into(),
            side: "S".into(),
            px: "78000".into(),
            sz: "0.005".into(),
            time: 2000,
            oid: None,
            fee: None,
        };
        svc.apply_fill_to_positions(&account, &fill2);

        assert!((svc.get_position_size(&account, "BTC").await - 0.005).abs() < 0.0001);
    }

    #[tokio::test]
    async fn test_service_wait_for_fills_timeout() {
        let svc = HydromancerFeedService::new(
            "ws://fake".into(),
            "http://fake".into(),
            "http://fake-hydromancer".into(),
            None,
            vec![],
        );
        let account = test_account();

        let fills = svc
            .wait_for_fills(&account, Duration::from_millis(100))
            .await;
        assert!(fills.is_empty());
    }

    #[tokio::test]
    async fn test_service_wait_for_fills_receives() {
        let svc = Arc::new(HydromancerFeedService::new(
            "ws://fake".into(),
            "http://fake".into(),
            "http://fake-hydromancer".into(),
            None,
            vec![],
        ));
        let account = test_account();

        let svc_clone = svc.clone();
        let account_clone = account;

        // Spawn a task that sends a fill after 50ms
        tokio::spawn(async move {
            tokio::time::sleep(Duration::from_millis(50)).await;
            let _ = svc_clone.event_tx.send(HydromancerEvent::Fill {
                account: account_clone,
                fill: FillEvent {
                    coin: "BTC".into(),
                    side: "B".into(),
                    px: "77000".into(),
                    sz: "0.01".into(),
                    time: 1000,
                    oid: None,
                    fee: None,
                },
            });
        });

        let fills = svc.wait_for_fills(&account, Duration::from_secs(1)).await;
        assert_eq!(fills.len(), 1);
        assert_eq!(fills[0].coin, "BTC");
        assert_eq!(fills[0].sz, "0.01");
    }

    #[tokio::test]
    async fn test_service_wait_ignores_other_accounts() {
        let svc = Arc::new(HydromancerFeedService::new(
            "ws://fake".into(),
            "http://fake".into(),
            "http://fake-hydromancer".into(),
            None,
            vec![],
        ));
        let account = test_account();
        let other = WalletAddress::from_str("0x0000000000000000000000000000000000000001").unwrap();

        let svc_clone = svc.clone();
        tokio::spawn(async move {
            tokio::time::sleep(Duration::from_millis(50)).await;
            let _ = svc_clone.event_tx.send(HydromancerEvent::Fill {
                account: other,
                fill: FillEvent {
                    coin: "ETH".into(),
                    side: "S".into(),
                    px: "2500".into(),
                    sz: "1.0".into(),
                    time: 1000,
                    oid: None,
                    fee: None,
                },
            });
        });

        let fills = svc
            .wait_for_fills(&account, Duration::from_millis(200))
            .await;
        assert!(fills.is_empty());
    }
}