fix(hyperliquid): dedup userFills to prevent position/PnL double-count

crates/bb-core/src/helpers/mod.rs

@@ -5,9 +5,11 @@
 pub mod client_id;
 pub mod inventory;
 pub mod parse;
+pub mod recent_ids;
 pub mod tick_feed;
 
 pub use client_id::ClientIdIssuer;
 pub use inventory::InventoryTracker;
 pub use parse::parse_decimal_or_warn;
+pub use recent_ids::RecentIds;
 pub use tick_feed::TickFeed;

crates/bb-core/src/helpers/recent_ids.rs

@@ -0,0 +1,72 @@
+//! Bounded set of recently-seen ids with FIFO eviction. Used by exchange
+//! adapters to drop fills replayed across a WS reconnect — emitting a fill
+//! twice would double-count the position — without growing memory unbounded.
+
+use std::collections::{HashSet, VecDeque};
+
+/// Bounded FIFO dedup set. `insert` returns `true` the first time an id is
+/// seen and `false` for a repeat, evicting the oldest id once `cap` is
+/// exceeded so memory stays bounded.
+#[derive(Debug, Clone)]
+pub struct RecentIds {
+    set: HashSet<String>,
+    order: VecDeque<String>,
+    cap: usize,
+}
+
+impl RecentIds {
+    /// `cap` is clamped to at least 1. A zero capacity would evict every id
+    /// the instant it is inserted — silently disabling dedup and reintroducing
+    /// the double-count it exists to prevent — so we coerce it, matching the
+    /// `.max(1)` guard `TickFeed` and `Volatility` use for degenerate sizes.
+    #[must_use]
+    pub fn new(cap: usize) -> Self {
+        Self { set: HashSet::new(), order: VecDeque::new(), cap: cap.max(1) }
+    }
+
+    /// Record `id`; returns `true` if it's new, `false` if already seen.
+    pub fn insert(&mut self, id: &str) -> bool {
+        if !self.set.insert(id.to_string()) {
+            return false;
+        }
+        self.order.push_back(id.to_string());
+        if self.order.len() > self.cap
+            && let Some(evicted) = self.order.pop_front()
+        {
+            self.set.remove(&evicted);
+        }
+        true
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::RecentIds;
+
+    #[test]
+    fn recent_ids_dedups_exact_repeats() {
+        let mut seen = RecentIds::new(4);
+        assert!(seen.insert("a"), "first sighting is new");
+        assert!(!seen.insert("a"), "exact repeat is a duplicate");
+        assert!(seen.insert("b"), "different id is new");
+    }
+
+    #[test]
+    fn recent_ids_zero_cap_is_clamped_and_still_dedups() {
+        // A zero capacity must not silently disable dedup — it's clamped to 1.
+        let mut seen = RecentIds::new(0);
+        assert!(seen.insert("a"), "first sighting is new");
+        assert!(!seen.insert("a"), "immediate repeat is still caught");
+    }
+
+    #[test]
+    fn recent_ids_evicts_oldest_past_cap() {
+        let mut seen = RecentIds::new(2);
+        seen.insert("a");
+        seen.insert("b");
+        seen.insert("c"); // evicts "a"
+        assert!(seen.insert("a"), "evicted id is treated as new again");
+        // memory stays bounded at the cap
+        assert!(seen.set.len() <= 2);
+    }
+}

crates/exchanges/bullet/src/connection.rs

@@ -13,13 +13,13 @@
 //!   - `OrderUpdateData::PlaceOrder` / `Cancel` emit only `OrderLifecycle` — they carry no
 //!     execution, so there's no `Trade` to emit.
 
-use std::collections::{HashSet, VecDeque};
 use std::sync::Arc;
 
 use bb_core::error::BotError;
 use bb_core::events::{BookUpdate, MarkPriceUpdate, OrderLifecycle, Trade};
 use bb_core::harness::MpscFeed;
 use bb_core::health::ConnectionHealth;
+use bb_core::helpers::RecentIds;
 use bullet_rust_sdk::ws::models::ServerMessage;
 use bullet_rust_sdk::{
     Client, ManagedWebsocket, Network, OrderbookDepth, Topic, UserActionDiscriminants, WsEvent,
@@ -38,34 +38,6 @@ const MARK_CHANNEL_CAPACITY: usize = 256;
 /// small recent window, so this is far more than enough while bounding memory.
 const MAX_SEEN_TRADE_IDS: usize = 8_192;
 
-/// Bounded set of recently-seen ids with FIFO eviction. Used to drop fills
-/// replayed across a reconnect (which would otherwise double-count the
-/// position) without growing memory without bound.
-struct RecentIds {
-    set: HashSet<String>,
-    order: VecDeque<String>,
-    cap: usize,
-}
-
-impl RecentIds {
-    fn new(cap: usize) -> Self {
-        Self { set: HashSet::new(), order: VecDeque::new(), cap }
-    }
-
-    /// Record `id`; returns `true` if it's new, `false` if already seen.
-    fn insert(&mut self, id: &str) -> bool {
-        if !self.set.insert(id.to_string()) {
-            return false;
-        }
-        self.order.push_back(id.to_string());
-        if self.order.len() > self.cap
-            && let Some(evicted) = self.order.pop_front()
-        {
-            self.set.remove(&evicted);
-        }
-        true
-    }
-}
 use crate::config::BulletConfig;
 use crate::convert;
 
@@ -261,27 +233,3 @@ async fn muxer_loop(
         }
     }
 }
-
-#[cfg(test)]
-mod tests {
-    use super::RecentIds;
-
-    #[test]
-    fn recent_ids_dedups_exact_repeats() {
-        let mut seen = RecentIds::new(4);
-        assert!(seen.insert("a"), "first sighting is new");
-        assert!(!seen.insert("a"), "exact repeat is a duplicate");
-        assert!(seen.insert("b"), "different id is new");
-    }
-
-    #[test]
-    fn recent_ids_evicts_oldest_past_cap() {
-        let mut seen = RecentIds::new(2);
-        seen.insert("a");
-        seen.insert("b");
-        seen.insert("c"); // evicts "a"
-        assert!(seen.insert("a"), "evicted id is treated as new again");
-        // memory stays bounded at the cap
-        assert!(seen.set.len() <= 2);
-    }
-}

crates/exchanges/hyperliquid/src/connection.rs

@@ -20,9 +20,10 @@ use bb_core::error::BotError;
 use bb_core::events::{BookUpdate, MarkPriceUpdate, OrderLifecycle, Trade};
 use bb_core::harness::MpscFeed;
 use bb_core::health::ConnectionHealth;
+use bb_core::helpers::RecentIds;
 use ethers::signers::{LocalWallet, Signer};
 use ethers::types::H160;
-use hyperliquid_rust_sdk::{BaseUrl, ExchangeClient, InfoClient, Message, Subscription};
+use hyperliquid_rust_sdk::{BaseUrl, ExchangeClient, InfoClient, Message, Subscription, TradeInfo};
 use tokio::sync::mpsc;
 
 use crate::broker::{ClientIdMap, HyperliquidBroker, new_client_id_map};
@@ -35,6 +36,11 @@ use crate::convert;
 const BOOK_CHANNEL_CAPACITY: usize = 4_096;
 const MARK_CHANNEL_CAPACITY: usize = 256;
 
+/// Cap on remembered fill ids for replay dedup — mirrors the Bullet adapter.
+/// A reconnect replays only a small recent window, so this bounds memory while
+/// being far more than enough.
+const MAX_SEEN_TRADE_IDS: usize = 8_192;
+
 /// HL's WS sends data continuously (`AllMids` ~250ms, `ActiveAssetCtx`, depth);
 /// a gap longer than this is treated as a transparent reconnect, triggering
 /// a reconcile signal so strategies can resync against REST.
@@ -170,6 +176,42 @@ pub async fn connect(
     Ok((broker, feeds))
 }
 
+/// Decide which `userFills` entries to emit as `Trade`s, updating dedup state.
+///
+/// Hyperliquid replays a historical fill snapshot on every (re)subscribe
+/// (`is_snapshot = true`). The *initial* snapshot after startup duplicates the
+/// REST `get_positions()` seed, so its fills are recorded (to suppress later
+/// duplicates) but not emitted. Every fill after that is emitted at most once,
+/// keyed on `trade_id` (`tid`): this drops reconnect-snapshot replays while
+/// still surfacing a genuinely new fill that landed during a disconnect gap
+/// (inventory is not otherwise re-seeded on reconnect).
+fn fills_to_emit(
+    fills: &[TradeInfo],
+    is_snapshot: bool,
+    fills_primed: &mut bool,
+    seen: &mut RecentIds,
+    client_ids: &ClientIdMap,
+    target_coin: &str,
+) -> Vec<Trade> {
+    let initial_snapshot = is_snapshot && !*fills_primed;
+    let mut out = Vec::new();
+    for fill in fills.iter().filter(|f| f.coin == target_coin) {
+        if let Some(trade) = convert::fill_to_trade(fill, client_ids) {
+            let first_time = match &trade.trade_id {
+                Some(id) => seen.insert(id),
+                None => true, // no id to dedup on — emit
+            };
+            if first_time && !initial_snapshot {
+                out.push(trade);
+            }
+        }
+    }
+    if is_snapshot {
+        *fills_primed = true;
+    }
+    out
+}
+
 /// Muxer task — reads the WS message stream, classifies each `Message`, and
 /// forwards converted events into the typed channels. Holds `ws_info` so the
 /// WS connection stays alive for the lifetime of the task.
@@ -196,6 +238,10 @@ async fn muxer_loop(
     // a reconnect — worth surfacing.
     let mut last_order_timestamp: u64 = 0;
     let mut last_msg_at = Instant::now();
+    // Dedup for `userFills`: HL replays a historical snapshot on every
+    // (re)subscribe, which would otherwise double-count the position.
+    let mut seen_fills = RecentIds::new(MAX_SEEN_TRADE_IDS);
+    let mut fills_primed = false;
     loop {
         let recv = tokio::time::timeout(HL_WS_QUIET_THRESHOLD, ws_rx.recv()).await;
         let msg = match recv {
@@ -254,10 +300,16 @@ async fn muxer_loop(
                 }
             }
             Message::UserFills(f) => {
-                for fill in f.data.fills.iter().filter(|f| f.coin == target_coin) {
-                    if let Some(trade) = convert::fill_to_trade(fill, &client_ids) {
-                        let _ = trade_tx.send(trade);
-                    }
+                let is_snapshot = f.data.is_snapshot.unwrap_or(false);
+                for trade in fills_to_emit(
+                    &f.data.fills,
+                    is_snapshot,
+                    &mut fills_primed,
+                    &mut seen_fills,
+                    &client_ids,
+                    &target_coin,
+                ) {
+                    let _ = trade_tx.send(trade);
                 }
             }
             Message::AllMids(m) => {
@@ -390,3 +442,93 @@ mod tests {
         assert!(!account_mode_is_unified("null"));
     }
 }
+
+#[cfg(test)]
+mod fill_dedup_tests {
+    use super::*;
+
+    fn fill(coin: &str, tid: u64) -> TradeInfo {
+        TradeInfo {
+            coin: coin.to_string(),
+            side: "B".to_string(),
+            px: "100".to_string(),
+            sz: "1".to_string(),
+            time: 0,
+            hash: String::new(),
+            start_position: "0".to_string(),
+            dir: "Open Long".to_string(),
+            closed_pnl: "0".to_string(),
+            oid: 1,
+            cloid: None,
+            crossed: false,
+            fee: "0".to_string(),
+            fee_token: "USDC".to_string(),
+            tid,
+        }
+    }
+
+    #[test]
+    fn initial_snapshot_is_recorded_but_not_emitted() {
+        let mut primed = false;
+        let mut seen = RecentIds::new(64);
+        let ids = new_client_id_map();
+        let emitted = fills_to_emit(
+            &[fill("BTC", 1), fill("BTC", 2)],
+            true,
+            &mut primed,
+            &mut seen,
+            &ids,
+            "BTC",
+        );
+        assert!(emitted.is_empty(), "initial snapshot fills must not be emitted");
+        assert!(primed, "snapshot marks the fill stream primed");
+        // The tids were recorded: a later live push of the same fill is dropped.
+        let again = fills_to_emit(&[fill("BTC", 1)], false, &mut primed, &mut seen, &ids, "BTC");
+        assert!(again.is_empty(), "already-seen tid is dropped");
+    }
+
+    #[test]
+    fn live_fill_emitted_exactly_once() {
+        let mut primed = true; // stream already primed past the initial snapshot
+        let mut seen = RecentIds::new(64);
+        let ids = new_client_id_map();
+        let first = fills_to_emit(&[fill("BTC", 10)], false, &mut primed, &mut seen, &ids, "BTC");
+        assert_eq!(first.len(), 1, "a new live fill is emitted once");
+        let dup = fills_to_emit(&[fill("BTC", 10)], false, &mut primed, &mut seen, &ids, "BTC");
+        assert!(dup.is_empty(), "duplicate live fill is dropped");
+    }
+
+    #[test]
+    fn reconnect_snapshot_emits_only_the_gap_fill() {
+        let mut primed = false;
+        let mut seen = RecentIds::new(64);
+        let ids = new_client_id_map();
+        // Initial snapshot: tids 1,2 recorded, not emitted.
+        fills_to_emit(&[fill("BTC", 1), fill("BTC", 2)], true, &mut primed, &mut seen, &ids, "BTC");
+        // Live fill tid 3.
+        assert_eq!(
+            fills_to_emit(&[fill("BTC", 3)], false, &mut primed, &mut seen, &ids, "BTC").len(),
+            1
+        );
+        // Reconnect snapshot replays 1,2,3 and carries a new gap fill tid 4.
+        let after = fills_to_emit(
+            &[fill("BTC", 1), fill("BTC", 2), fill("BTC", 3), fill("BTC", 4)],
+            true,
+            &mut primed,
+            &mut seen,
+            &ids,
+            "BTC",
+        );
+        assert_eq!(after.len(), 1, "only the new gap fill is emitted on reconnect");
+        assert_eq!(after[0].trade_id.as_deref(), Some("4"));
+    }
+
+    #[test]
+    fn fills_for_other_coins_are_ignored() {
+        let mut primed = true;
+        let mut seen = RecentIds::new(64);
+        let ids = new_client_id_map();
+        let emitted = fills_to_emit(&[fill("ETH", 5)], false, &mut primed, &mut seen, &ids, "BTC");
+        assert!(emitted.is_empty(), "fills for a non-target coin are filtered out");
+    }
+}