From 28e8c3dd0f7663fa551d3afe9d98847a7c44ac4d Mon Sep 17 00:00:00 2001 From: TAMARA LIPOWSKI Date: Mon, 3 Aug 2026 17:30:01 -0400 Subject: [PATCH] fix(token-prices): price a token from the median of its deepest paths Token prices came from the candidate path with the tightest round-trip spread. Spread only measures whether a probe-sized round trip returns symmetric; it says nothing about whether the rate is right. A pool holding one side and almost none of the other quotes a wrong rate with a tight spread and deep liquidity, and on BSC such a WBNB-USDT pool priced USDT 12191x too low. No threshold on that pool alone can catch it, because its reserves are internally consistent. What catches it is that no other path agrees. Candidates are now ranked by the depth of their thinnest hop and the deepest PRICED_PATHS_PER_TOKEN are simulated. From three quotes up the token's price is their median, so a lone disagreeing pool is outvoted; below three it is the deepest path's, since two quotes cannot outvote each other. Depth replaces spread as the ranking key because the wrong pool wins on spread, so ranking by spread would fill the sample with it. Depth ranks paths rather than admitting them. A hop with no depth recorded reads as zero and sorts last instead of being dropped, so a token whose pools are all thin keeps a rough price rather than losing it. Token prices now require pool_depths, which reorders the derived schedule to spot prices, then depths, then token prices. Co-Authored-By: Claude Opus 5 (1M context) --- .../derived/computations/token_gas_price.rs | 580 ++++++++++++++---- fynd-core/src/derived/manager.rs | 9 +- 2 files changed, 449 insertions(+), 140 deletions(-) diff --git a/fynd-core/src/derived/computations/token_gas_price.rs b/fynd-core/src/derived/computations/token_gas_price.rs index d5fc880b9..5dd43f9c9 100644 --- a/fynd-core/src/derived/computations/token_gas_price.rs +++ b/fynd-core/src/derived/computations/token_gas_price.rs @@ -1,18 +1,19 @@ -//! Computes the `mid_price` of tokens relative to a gas token (e.g., ETH), selecting paths -//! by the lowest spread (the most reliable price) derived from full simulation of both buy and sell -//! directions. +//! Computes the `mid_price` of tokens relative to a gas token (e.g., ETH) as the median of what a +//! token's deepest paths quote, each from a full simulation of both buy and sell directions. //! //! # Algorithm //! -//! 1. **Path Discovery (DFS)**: Enumerate all paths from gas_token to each reachable token, scoring -//! by spot-price spread: `|forward_spot - 1/reverse_spot|`. Lower spread = better score. +//! 1. **Path Discovery (DFS)**: Enumerate all paths from gas_token to each reachable token, keeping +//! the depth of each path's thinnest hop. //! -//! 2. **Sort**: Order paths per token by spread score (lowest spread first). +//! 2. **Sort**: Order paths per token by depth, deepest first, and keep the first +//! `PRICED_PATHS_PER_TOKEN`. //! -//! 3. **Round-Robin Simulation**: For each token, simulate paths in ranked order and compute their -//! spread and mid_price by simulating both directions on the same path. Pick the path with the -//! tightest spread for each token, as this indicates the most reliable/liquid route, and provide -//! its mid_price as the token's price. +//! 3. **Round-Robin Simulation**: For each token, simulate every kept path. From three quotes up +//! the token's price is their median, so a pool quoting a rate no other path agrees with is +//! outvoted; below three it is the deepest path's. Spread ranking could do neither: a pool +//! holding one side and almost none of the other quotes a wrong rate with a tight spread and +//! deep liquidity, and nothing about that pool on its own says so. //! //! # Price Formulas //! @@ -26,9 +27,8 @@ //! //! # Dependencies //! -//! This computation depends on [`SpotPrices`](crate::derived::types::SpotPrices) being -//! available in the [`DerivedData`](crate::derived::store::DerivedData). -//! Ensure `SpotPriceComputation` runs before this computation. +//! Needs `SpotPrices` and `PoolDepths` in `DerivedData`, so `SpotPriceComputation` and +//! `PoolDepthComputation` must run first. use std::collections::{HashMap, HashSet}; @@ -47,11 +47,14 @@ use crate::{ ComputationId, ComputationOutput, ComputationRequirements, DerivedComputation, FailedItem, FailedItemError, }, - computations::spot_price::SpotPriceComputation, + computations::{pool_depth::PoolDepthComputation, spot_price::SpotPriceComputation}, error::ComputationError, manager::{ChangedComponents, SharedDerivedDataRef}, store::DerivedData, - types::{SpotPriceKey, SpotPrices, TokenGasPrices, TokenPriceEntry, TokenPricesWithDeps}, + types::{ + PoolDepthKey, PoolDepths, SpotPriceKey, SpotPrices, TokenGasPrices, TokenPriceEntry, + TokenPricesWithDeps, + }, }, feed::market_data::{MarketData, MarketState}, graph::{GraphManager, Path, PetgraphStableDiGraphManager}, @@ -59,18 +62,47 @@ use crate::{ MostLiquidAlgorithm, }; -/// A path with its score +/// A candidate path with the depth of its thinnest hop, in gas-token units. #[derive(Clone)] struct CandidatePath<'a> { path: Path<'a, ()>, - score: f64, + depth: f64, +} + +/// What one simulated path returned, and the two numbers used to choose between paths. +struct Quote { + price: Price, + /// `price` as a float, for ordering quotes by what they say the token is worth. + ratio: f64, + /// Depth of the path's thinnest hop, in gas-token units. + depth: f64, + components: HashSet, +} + +/// A price as a float, for ordering quotes against each other. Non-finite when the denominator is +/// zero or either side overflows f64, which keeps the quote out of the median. +fn ratio_of(price: &Price) -> f64 { + let numerator = price + .numerator + .to_f64() + .unwrap_or(f64::NAN); + let denominator = price + .denominator + .to_f64() + .unwrap_or(f64::NAN); + numerator / denominator } -/// Computes token prices relative to the gas token. Returns the buy price for the path -/// with the lowest spread (most reliable) that we managed to find. +/// How many quotes a token's price is chosen from. That is quotes, not attempts: a path whose +/// simulation fails does not count against it, so a token is not left unpriced by having several +/// deep paths that do not simulate. It is also the number of independent quotes a single wrong pool +/// has to outvote. +const PRICED_PATHS_PER_TOKEN: usize = 5; + +/// Computes token prices relative to the gas token from what its deepest paths quote. /// -/// Uses DFS to discover paths, spot prices for ranking, and full simulation -/// for accurate output amounts and spread calculation. +/// Uses DFS to discover paths, pool depths for ranking, and full simulation for accurate output +/// amounts. #[derive(Debug, Clone)] pub struct TokenGasPriceComputation { /// The gas token address (e.g., ETH). @@ -107,11 +139,16 @@ impl TokenGasPriceComputation { Self { gas_token, ..self } } - /// DFS to discover all paths from gas_token, scored by spot-price spread. + /// DFS to discover all paths from gas_token, each carrying the depth of its thinnest hop. + /// + /// Every path a token has is kept. Depth ranks them rather than admitting them, so a token + /// whose pools are all thin still gets a price — a rough one beats none, and the price of a + /// token nothing much trades against cannot be sharp anyway. fn discover_paths<'a>( &self, graph_manager: &'a PetgraphStableDiGraphManager<()>, spot_prices: &SpotPrices, + pool_depths: &PoolDepths, ) -> Result>>, ComputationError> { let graph = graph_manager.graph(); @@ -127,14 +164,15 @@ impl TokenGasPriceComputation { token_node: NodeIndex, path: Path<'a, ()>, forward_spot: f64, - reverse_spot: f64, + /// Depth of the thinnest hop taken so far, in gas-token units. + depth: f64, } let mut stack = vec![DfsFrame { token_node: entry_node, path: Path::new(), forward_spot: 1.0, - reverse_spot: 1.0, + depth: f64::INFINITY, }]; while let Some(frame) = stack.pop() { @@ -143,19 +181,10 @@ impl TokenGasPriceComputation { // Record non-empty paths (skip the starting node's empty path) if !frame.path.is_empty() { - // Compute spread from spot prices: - // buy_price = forward_spot (target per gas when buying) - // sell_price = 1/reverse_spot (target per gas when selling) - // spread = |buy_price - sell_price| - // Score = spread directly (lower = better, 0 for symmetric pools) - let buy_price = frame.forward_spot; - let sell_price = 1.0 / frame.reverse_spot; - let spot_spread = (buy_price - sell_price).abs(); - paths_by_token .entry(token_reached.clone()) .or_default() - .push(CandidatePath { path: frame.path.clone(), score: spot_spread }); + .push(CandidatePath { path: frame.path.clone(), depth: frame.depth }); } // Stop exploring further if max depth reached @@ -183,7 +212,17 @@ impl TokenGasPriceComputation { let Some(&fwd_spot) = spot_prices.get(&fwd_key) else { continue; }; - let Some(&rev_spot) = spot_prices.get(&rev_key) else { + if !spot_prices.contains_key(&rev_key) { + continue; + } + + let Some(hop_depth) = Self::hop_depth_in_gas_units( + pool_depths, + &component_id, + token_reached, + next_token, + frame.forward_spot, + ) else { continue; }; @@ -191,7 +230,7 @@ impl TokenGasPriceComputation { token_node: next_node, path: new_path, forward_spot: frame.forward_spot * fwd_spot, - reverse_spot: frame.reverse_spot * rev_spot, + depth: frame.depth.min(hop_depth), }); } } @@ -199,6 +238,50 @@ impl TokenGasPriceComputation { Ok(paths_by_token) } + /// A hop's depth in gas-token units, or `None` when it has none recorded or the conversion has + /// nothing to work from. + /// + /// `PoolDepthComputation` reports depth as the largest input a pool takes before its price + /// moves further than the slippage threshold, measured in `token_in`. + /// `gas_to_token_in_spot` is how many `token_in` one gas token buys, so dividing converts + /// the depth into gas-token units. + /// + /// A hop whose depth never computed reads as `None` rather than as deep: the alternative is + /// pricing off a pool whose liquidity is unknown. + fn hop_depth_in_gas_units( + pool_depths: &PoolDepths, + component_id: &ComponentId, + token_in: &Address, + token_out: &Address, + gas_to_token_in_spot: f64, + ) -> Option { + if !gas_to_token_in_spot.is_finite() || gas_to_token_in_spot <= 0.0 { + return None; + } + let key: PoolDepthKey = (component_id.clone(), token_in.clone(), token_out.clone()); + let depth = pool_depths + .get(&key) + .and_then(ToPrimitive::to_f64) + .unwrap_or(0.0); + Some(depth / gas_to_token_in_spot) + } + + /// The quote a token's price comes from: the median by price once three paths have quoted, + /// and the deepest path before that. + /// + /// Two quotes cannot outvote each other, so depth decides instead of an arbitrary middle. + /// Either way the price is one a path actually returned rather than an average of several, + /// so it keeps exact numerator and denominator along with the components it came from. + fn chosen_quote(mut quotes: Vec) -> Option { + if quotes.len() < 3 { + quotes.sort_by(|a, b| a.depth.total_cmp(&b.depth)); + return quotes.pop(); + } + quotes.sort_by(|a, b| a.ratio.total_cmp(&b.ratio)); + let middle = quotes.len() / 2; + quotes.drain(middle..=middle).next() + } + /// Compute the spread and mid_price for a given path by simulating both directions. /// /// Returns (spread_ratio, mid_price, path_components) where: @@ -315,6 +398,7 @@ impl TokenGasPriceComputation { /// /// * `market`: The market data to simulate token prices on. /// * `spot_prices`: The spot prices to use for the simulation. + /// * `pool_depths`: The depths that decide which hops are deep enough to price through. /// * `filter_tokens`: An optional set of tokens to filter the simulation by. If None, all /// tokens are simulated. /// @@ -326,9 +410,10 @@ impl TokenGasPriceComputation { &self, market: &MarketData, spot_prices: &SpotPrices, + pool_depths: &PoolDepths, filter_tokens: Option<&HashSet
>, ) -> Result< - (HashMap)>, u64, Vec), + (HashMap)>, u64, Vec), ComputationError, > { // Brief lock 1: topology + gas_price + block (all cheap clones) @@ -350,7 +435,7 @@ impl TokenGasPriceComputation { let needed_component_ids = { let mut graph_manager = PetgraphStableDiGraphManager::new(); graph_manager.initialize_graph(&topology); - let mut paths = self.discover_paths(&graph_manager, spot_prices)?; + let mut paths = self.discover_paths(&graph_manager, spot_prices, pool_depths)?; if let Some(tokens) = filter_tokens { paths.retain(|token, _| tokens.contains(token)); } @@ -377,7 +462,7 @@ impl TokenGasPriceComputation { // Rediscover paths from subset + simulate (no lock, expensive EVM simulation) let mut graph_manager = PetgraphStableDiGraphManager::new(); graph_manager.initialize_graph(&subset.component_topology()); - let mut paths_by_token = self.discover_paths(&graph_manager, spot_prices)?; + let mut paths_by_token = self.discover_paths(&graph_manager, spot_prices, pool_depths)?; // Optionally filter to only requested tokens if let Some(tokens) = filter_tokens { @@ -403,50 +488,63 @@ impl TokenGasPriceComputation { }) .collect(); - // Sort each token's paths: lowest spread last (for popping). A NaN score (degenerate - // pool math in the spread computation) cannot rank a path and would panic a - // partial_cmp-based sort, so drop those candidates and sort with the float total order. + // Order each token's candidates shallowest first, so popping takes the deepest. A + // non-finite depth cannot rank a path and would panic a partial_cmp-based sort, so drop + // those candidates and sort with the float total order. for paths in paths_by_token.values_mut() { - paths.retain(|path| !path.score.is_nan()); - paths.sort_by(|a, b| b.score.total_cmp(&a.score)); + paths.retain(|path| path.depth.is_finite()); + paths.sort_by(|a, b| a.depth.total_cmp(&b.depth)); } - // Round-robin: pop one candidate per token each round, keep best by spread - let mut best_prices: HashMap)> = HashMap::new(); + // Round-robin: pop one candidate per token each round, collecting every quote that + // simulates, then reduce each token's quotes to their median. + let mut quotes: HashMap> = HashMap::new(); let mut candidates_exhausted = false; while !candidates_exhausted { candidates_exhausted = true; for (token, candidate_paths) in paths_by_token.iter_mut() { + // Enough quotes for this token; drop its remaining candidates so the loop ends. + if quotes.get(token).map_or(0, Vec::len) >= PRICED_PATHS_PER_TOKEN { + candidate_paths.clear(); + continue; + } let Some(candidate) = candidate_paths.pop() else { continue; }; candidates_exhausted = false; - match self.compute_spread_and_mid_price(candidate.path, &subset, &gas_price) { - Ok((spread, price, components)) => { - let is_better = best_prices - .get(token) - .map(|(existing_spread, _, _)| spread < *existing_spread) - .unwrap_or(true); - if is_better { - trace!( - token = ?token, - spread_ratio = spread, - "found better price (lower spread)" - ); - best_prices.insert(token.clone(), (spread, price, components)); - } - } - Err(_) => continue, + // A non-finite spread means the round trip was degenerate, so the price it came + // with does not belong in the median. + let depth = candidate.depth; + let Ok((spread, price, components)) = + self.compute_spread_and_mid_price(candidate.path, &subset, &gas_price) + else { + continue; + }; + let ratio = ratio_of(&price); + if spread.is_finite() && ratio.is_finite() { + quotes + .entry(token.clone()) + .or_default() + .push(Quote { price, ratio, depth, components }); } } } + let mut best_prices: HashMap)> = HashMap::new(); + for (token, token_quotes) in quotes { + let priced_paths = token_quotes.len(); + if let Some(quote) = Self::chosen_quote(token_quotes) { + trace!(token = ?token, priced_paths, "chose from the token's priced paths"); + best_prices.insert(token, (quote.price, quote.components)); + } + } + // Extend each token's path_components with all candidate path components so // incremental recomputation fires when any competing path's pool changes. - for (token, (_, _, components)) in best_prices.iter_mut() { + for (token, (_, components)) in best_prices.iter_mut() { if let Some(all_comps) = all_candidate_components.get(token) { components.extend(all_comps.iter().cloned()); } @@ -478,7 +576,7 @@ impl TokenGasPriceComputation { changed: &ChangedComponents, ) -> Result>, ComputationError> { // Read all needed data from store in a single lock acquisition. - let (existing_deps, existing_prices, spot_prices) = { + let (existing_deps, existing_prices, spot_prices, pool_depths) = { let store_guard = store.read().await; // Need existing deps to do incremental computation. @@ -492,8 +590,12 @@ impl TokenGasPriceComputation { .spot_prices() .ok_or(ComputationError::MissingDependency("spot_prices"))? .clone(); + let pool_depths = store_guard + .pool_depths() + .ok_or(ComputationError::MissingDependency("pool_depths"))? + .clone(); - (existing_deps, existing_prices, spot_prices) + (existing_deps, existing_prices, spot_prices, pool_depths) }; let changed_components = changed.all_changed_ids(); @@ -520,7 +622,7 @@ impl TokenGasPriceComputation { ); let (best_prices, block, _) = self - .simulate_token_prices(market, &spot_prices, Some(&tokens_to_recompute)) + .simulate_token_prices(market, &spot_prices, &pool_depths, Some(&tokens_to_recompute)) .await?; // Merge results into existing prices and deps @@ -529,7 +631,7 @@ impl TokenGasPriceComputation { let mut failed_items: Vec = Vec::new(); for token in &tokens_to_recompute { - if let Some((_, price, components)) = best_prices.get(token) { + if let Some((price, components)) = best_prices.get(token) { new_deps.insert( token.clone(), TokenPriceEntry { price: price.clone(), path_components: components.clone() }, @@ -562,7 +664,7 @@ impl DerivedComputation for TokenGasPriceComputation { const ID: ComputationId = "token_prices"; fn requirements(&self) -> ComputationRequirements { - ComputationRequirements::fresh([SpotPriceComputation::ID]) + ComputationRequirements::fresh([SpotPriceComputation::ID, PoolDepthComputation::ID]) } fn persist( @@ -594,23 +696,29 @@ impl DerivedComputation for TokenGasPriceComputation { // Fall through to full compute if incremental is not possible } - // Read spot prices from store (independent of market lock). - let spot_prices = store - .read() - .await - .spot_prices() - .ok_or(ComputationError::MissingDependency("spot_prices"))? - .clone(); + // Read spot prices and depths from store (independent of market lock). + let (spot_prices, pool_depths) = { + let store_guard = store.read().await; + let spot_prices = store_guard + .spot_prices() + .ok_or(ComputationError::MissingDependency("spot_prices"))? + .clone(); + let pool_depths = store_guard + .pool_depths() + .ok_or(ComputationError::MissingDependency("pool_depths"))? + .clone(); + (spot_prices, pool_depths) + }; let (best_prices, block, failed_items) = self - .simulate_token_prices(market, &spot_prices, None) + .simulate_token_prices(market, &spot_prices, &pool_depths, None) .await?; // Build token prices with dependencies for incremental computation let mut token_prices_with_deps = TokenPricesWithDeps::new(); let mut token_prices = TokenGasPrices::new(); - for (token, (_, price, path_components)) in best_prices { + for (token, (price, path_components)) in best_prices { token_prices_with_deps .insert(token.clone(), TokenPriceEntry { price: price.clone(), path_components }); token_prices.insert(token, price); @@ -661,7 +769,13 @@ mod tests { // ==================== Test Helpers ==================== - /// Sets up a complete test environment: market with pools + precomputed spot prices. + /// Depth written for every pool direction by `setup_test_env`, ten times the probe so a test + /// that says nothing about depth prices as it always did. Tests about the depth check itself + /// call `set_depths` to overwrite specific directions. + const DEEP: u64 = 10_000_000_000_000_000_000; + + /// Sets up a complete test environment: market with pools + precomputed spot prices, and a + /// depth above the probe for every pool direction. /// Returns (market_guard, store) ready for computation. async fn setup_test_env( pools: Vec<(&str, &Token, &Token, MockProtocolSim)>, @@ -685,30 +799,63 @@ mod tests { .compute(&wrapped_market, &wrapped_store, &changed) .await .expect("spot price computation should succeed"); - wrapped_store - .try_write() - .unwrap() - .set_spot_prices(spot_prices_output.data, vec![], 0, true); + let mut pool_depths = PoolDepths::new(); + for (id, t1, t2, _) in &pools { + pool_depths.insert( + (id.to_string(), t1.address.clone(), t2.address.clone()), + BigUint::from(DEEP), + ); + pool_depths.insert( + (id.to_string(), t2.address.clone(), t1.address.clone()), + BigUint::from(DEEP), + ); + } + + { + let mut store_guard = wrapped_store.try_write().unwrap(); + store_guard.set_spot_prices(spot_prices_output.data, vec![], 0, true); + store_guard.set_pool_depths(pool_depths, vec![], 0, true); + } (wrapped_market, wrapped_store) } - async fn setup_graph_and_spot_prices( + /// A tenth of the probe, for the hop a test wants path discovery to skip. + const BELOW_PROBE: u64 = 100_000_000_000_000_000; + + /// Overwrites the depth of one pool direction, for tests that drive the depth check. + fn set_depths( + store: &SharedDerivedDataRef, + depths: Vec<(&str, &Address, &Address, u64)>, + ) -> PoolDepths { + let mut store_guard = store.try_write().unwrap(); + let mut pool_depths = store_guard + .pool_depths() + .cloned() + .unwrap_or_default(); + for (id, token_in, token_out, depth) in depths { + pool_depths.insert( + (id.to_string(), token_in.clone(), token_out.clone()), + BigUint::from(depth), + ); + } + store_guard.set_pool_depths(pool_depths.clone(), vec![], 0, true); + pool_depths + } + + async fn setup_graph_and_derived( pools: Vec<(&str, &Token, &Token, MockProtocolSim)>, - ) -> (PetgraphStableDiGraphManager<()>, SpotPrices) { + ) -> (PetgraphStableDiGraphManager<()>, SpotPrices, PoolDepths) { let (market, derived) = setup_test_env(pools).await; let market = market_read(&market); let mut graph = PetgraphStableDiGraphManager::new(); graph.initialize_graph(&market.component_topology()); - let spot_prices = derived - .try_write() - .unwrap() - .spot_prices() - .unwrap() - .clone(); - (graph, spot_prices) + let guard = derived.try_write().unwrap(); + let spot_prices = guard.spot_prices().unwrap().clone(); + let pool_depths = guard.pool_depths().unwrap().clone(); + (graph, spot_prices, pool_depths) } /// Creates a computation configured for the given gas token with standard settings. @@ -723,13 +870,13 @@ mod tests { let eth = token(0, "ETH"); let usdc = token(1, "USDC"); - let (graph_manager, spot_prices) = - setup_graph_and_spot_prices(vec![("pool", ð, &usdc, MockProtocolSim::new(2000.0))]) + let (graph_manager, spot_prices, pool_depths) = + setup_graph_and_derived(vec![("pool", ð, &usdc, MockProtocolSim::new(2000.0))]) .await; let computation = computation_for(ð.address); let paths = computation - .discover_paths(&graph_manager, &spot_prices) + .discover_paths(&graph_manager, &spot_prices, &pool_depths) .unwrap(); // Exactly 1 path to USDC (single hop via "pool") @@ -740,8 +887,9 @@ mod tests { assert_eq!(path.path.len(), 1, "path should be single hop"); assert_eq!(path.path.edge_data[0].component_id, "pool"); - // For a symmetric pool, spread = 0 - assert_eq!(path.score, 0.0); + // Depth is the thinnest hop's, converted to gas-token units; a single hop out of the gas + // token needs no conversion, so it is what setup_test_env wrote. + assert_eq!(path.depth, DEEP as f64); } #[tokio::test] @@ -750,7 +898,7 @@ mod tests { let mid = token(2, "MID"); let target = token(3, "TARGET"); - let (graph, spot_prices) = setup_graph_and_spot_prices(vec![ + let (graph, spot_prices, pool_depths) = setup_graph_and_derived(vec![ ("hop1", ð, &mid, MockProtocolSim::new(2.0)), ("hop2", &mid, &target, MockProtocolSim::new(3.0)), ]) @@ -758,7 +906,7 @@ mod tests { let computation = computation_for(ð.address); let paths = computation - .discover_paths(&graph, &spot_prices) + .discover_paths(&graph, &spot_prices, &pool_depths) .unwrap(); // MID: exactly 1 path (1-hop via hop1) @@ -766,7 +914,7 @@ mod tests { assert_eq!(mid_paths.len(), 1, "should have exactly 1 path to MID"); assert_eq!(mid_paths[0].path.len(), 1, "MID path should be 1 hop"); assert_eq!(mid_paths[0].path.edge_data[0].component_id, "hop1"); - assert_eq!(mid_paths[0].score, 0.0); + assert_eq!(mid_paths[0].depth, DEEP as f64); // TARGET: exactly 1 path (2-hop via hop1 → hop2) let target_paths = &paths[&target.address]; @@ -774,33 +922,88 @@ mod tests { assert_eq!(target_paths[0].path.len(), 2, "TARGET path should be 2 hops"); assert_eq!(target_paths[0].path.edge_data[0].component_id, "hop1"); assert_eq!(target_paths[0].path.edge_data[1].component_id, "hop2"); - assert_eq!(target_paths[0].score, 0.0); + // hop2's depth is in MID, worth 2 gas units each, so it is the thinner of the two. + assert_eq!(target_paths[0].depth, DEEP as f64 / 2.0); } #[tokio::test] - async fn nan_scored_paths_are_dropped_not_panicked_on() { - // Degenerate pool math can yield a NaN spot-price spread. A NaN-scored candidate must - // be dropped — never panicked on — and the affected token simply gets no price. + async fn non_finite_spot_prices_cannot_panic_or_reach_the_median() { + // Degenerate pool math can yield NaN spot prices. Ranking must not panic on them, and a hop + // whose depth cannot be converted out of its input token must not be walked: converting + // needs the spot product from the gas token, and NaN carries no amount. let eth = token(0, "ETH"); let usdc = token(1, "USDC"); + let dai = token(2, "DAI"); - let (market, _) = - setup_market_weighted(vec![("nan_pool", ð, &usdc, MockProtocolSim::new(2000.0))]); - // Inject NaN spot prices directly: the spread |forward - 1/reverse| becomes NaN. + let (market, _) = setup_market_weighted(vec![ + ("nan_pool", ð, &usdc, MockProtocolSim::new(2000.0)), + ("usdc_dai", &usdc, &dai, MockProtocolSim::new(1.0)), + ]); let mut spot_prices = SpotPrices::default(); - spot_prices - .insert(("nan_pool".to_string(), eth.address.clone(), usdc.address.clone()), f64::NAN); - spot_prices - .insert(("nan_pool".to_string(), usdc.address.clone(), eth.address.clone()), f64::NAN); + for (component, from, to) in [ + ("nan_pool", ð.address, &usdc.address), + ("nan_pool", &usdc.address, ð.address), + ("usdc_dai", &usdc.address, &dai.address), + ("usdc_dai", &dai.address, &usdc.address), + ] { + spot_prices.insert((component.to_string(), from.clone(), to.clone()), f64::NAN); + } + let mut pool_depths = PoolDepths::new(); + for (component, from, to) in + [("nan_pool", ð.address, &usdc.address), ("usdc_dai", &usdc.address, &dai.address)] + { + pool_depths + .insert((component.to_string(), from.clone(), to.clone()), BigUint::from(DEEP)); + } let computation = computation_for(ð.address); let (prices, _, _) = computation - .simulate_token_prices(&market, &spot_prices, None) + .simulate_token_prices(&market, &spot_prices, &pool_depths, None) .await - .expect("a NaN-scored path must not fail the computation"); + .expect("NaN spot prices must not fail the computation"); + + // USDC is one hop from the gas token, so its depth needs no conversion and it still prices. + assert!(prices.contains_key(&usdc.address), "the first hop needs no spot product"); + // DAI's hop is denominated in USDC, and the product that would convert it is NaN. + assert!(!prices.contains_key(&dai.address), "a NaN conversion must not yield a price"); + } + + #[tokio::test] + async fn test_compute_outvotes_the_bsc_pool_that_mispriced_usdt() { + // The case this change was written for, at its measured rates. On BSC a WBNB-USDT pool + // quoting 0.0464 USDT per BNB priced USDT 12191x under the 565.21 that USDC and BUSD + // independently agreed on (live monitor, 2026-07-29). The pool holds one side and almost + // none of the other, so it quotes that rate with deep liquidity and a tight round trip: + // neither depth nor spread rules it out. Only the disagreement does. + const AGREED: f64 = 565.21; + const LOPSIDED: f64 = AGREED / 12191.0; + + let wbnb = token(0, "WBNB"); + let usdt = token(1, "USDT"); + + let (market, derived) = setup_test_env(vec![ + ("pancake_v2", &wbnb, &usdt, MockProtocolSim::new(AGREED).with_fee(0.0025)), + ("pancake_v3", &wbnb, &usdt, MockProtocolSim::new(AGREED).with_fee(0.0005)), + ("biswap", &wbnb, &usdt, MockProtocolSim::new(AGREED).with_fee(0.003)), + ("lopsided", &wbnb, &usdt, MockProtocolSim::new(LOPSIDED)), + ]) + .await; + + let computation = computation_for(&wbnb.address); + let prices = computation + .compute(&market, &derived, &ChangedComponents::default()) + .await + .unwrap() + .data; + + let usdt_price = prices + .get(&usdt.address) + .expect("USDT should have price"); + let ratio = + usdt_price.numerator.to_f64().unwrap() / usdt_price.denominator.to_f64().unwrap(); assert!( - !prices.contains_key(&usdc.address), - "the NaN-scored path must be dropped, not selected" + (550.0..580.0).contains(&ratio), + "the three agreeing pools should outvote the lopsided one (~{AGREED}), got {ratio}" ); } @@ -811,7 +1014,7 @@ mod tests { let b = token(3, "B"); let c = token(4, "C"); - let (graph, spot_prices) = setup_graph_and_spot_prices(vec![ + let (graph, spot_prices, pool_depths) = setup_graph_and_derived(vec![ ("eth_a", ð, &a, MockProtocolSim::new(2.0)), ("a_b", &a, &b, MockProtocolSim::new(2.0)), ("b_c", &b, &c, MockProtocolSim::new(2.0)), @@ -821,7 +1024,7 @@ mod tests { // max_hops = 2 let computation = computation_for(ð.address); let paths = computation - .discover_paths(&graph, &spot_prices) + .discover_paths(&graph, &spot_prices, &pool_depths) .unwrap(); // A: exactly 1 path (1 hop via eth_a) @@ -829,7 +1032,6 @@ mod tests { assert_eq!(a_paths.len(), 1, "should have exactly 1 path to A"); assert_eq!(a_paths[0].path.len(), 1, "A path should be 1 hop"); assert_eq!(a_paths[0].path.edge_data[0].component_id, "eth_a"); - assert_eq!(a_paths[0].score, 0.0); // B: exactly 1 path (2 hops via eth_a → a_b) let b_paths = &paths[&b.address]; @@ -837,7 +1039,6 @@ mod tests { assert_eq!(b_paths[0].path.len(), 2, "B path should be 2 hops"); assert_eq!(b_paths[0].path.edge_data[0].component_id, "eth_a"); assert_eq!(b_paths[0].path.edge_data[1].component_id, "a_b"); - assert_eq!(b_paths[0].score, 0.0); // C: not reachable (would require 3 hops, exceeds max_hops=2) assert!(!paths.contains_key(&c.address), "C should NOT be reachable (3 hops)"); @@ -849,7 +1050,7 @@ mod tests { let usdc = token(1, "USDC"); // Two pools with different spot prices - let (graph, spot_prices) = setup_graph_and_spot_prices(vec![ + let (graph, spot_prices, pool_depths) = setup_graph_and_derived(vec![ ("pool_low", ð, &usdc, MockProtocolSim::new(1000.0)), ("pool_high", ð, &usdc, MockProtocolSim::new(2000.0)), ]) @@ -857,19 +1058,15 @@ mod tests { let computation = computation_for(ð.address); let paths = computation - .discover_paths(&graph, &spot_prices) + .discover_paths(&graph, &spot_prices, &pool_depths) .unwrap(); // Exactly 2 paths to USDC (one via each pool) let usdc_paths = &paths[&usdc.address]; assert_eq!(usdc_paths.len(), 2, "should have exactly 2 paths to USDC"); - // MockProtocolSim's spot_price is symmetric: forward_spot = 1/reverse_spot, - // so spread = |forward - 1/reverse| = 0 for all pools. - // TODO: Test with asymmetric simulation component to verify non-zero spread ranking. for path in usdc_paths { assert_eq!(path.path.len(), 1, "path should be single hop"); - assert_eq!(path.score, 0.0, "symmetric mock produces zero spread"); } // Verify both pools are discovered (order is arbitrary when scores are equal) @@ -885,6 +1082,78 @@ mod tests { assert!(component_ids.contains(&"pool_high")); } + #[tokio::test] + async fn test_discover_paths_records_the_thinnest_hop_depth() { + // Two pools price the same pair. Both are symmetric, so spread cannot tell them apart. + // Each path carries its own depth, which is what ranks them later. + let eth = token(0, "ETH"); + let usdc = token(1, "USDC"); + + let (market, derived) = setup_test_env(vec![ + ("deep", ð, &usdc, MockProtocolSim::new(2000.0)), + ("shallow", ð, &usdc, MockProtocolSim::new(1.0)), + ]) + .await; + let pool_depths = + set_depths(&derived, vec![("shallow", ð.address, &usdc.address, BELOW_PROBE)]); + let spot_prices = derived + .try_write() + .unwrap() + .spot_prices() + .unwrap() + .clone(); + + let mut graph = PetgraphStableDiGraphManager::new(); + graph.initialize_graph(&market_read(&market).component_topology()); + + let computation = computation_for(ð.address); + let paths = computation + .discover_paths(&graph, &spot_prices, &pool_depths) + .unwrap(); + + let mut usdc_paths: Vec<(&str, f64)> = paths[&usdc.address] + .iter() + .map(|c| { + ( + c.path.edge_data[0] + .component_id + .as_str(), + c.depth, + ) + }) + .collect(); + usdc_paths.sort_by(|a, b| a.1.total_cmp(&b.1)); + assert_eq!(usdc_paths, vec![("shallow", BELOW_PROBE as f64), ("deep", DEEP as f64)]); + } + + #[tokio::test] + async fn test_discover_paths_reads_an_unmeasured_hop_as_zero_depth() { + // A pool whose depth never computed still yields a path, at a depth that ranks it last. + let eth = token(0, "ETH"); + let usdc = token(1, "USDC"); + + let (market, derived) = + setup_test_env(vec![("eth_usdc", ð, &usdc, MockProtocolSim::new(2000.0))]).await; + let spot_prices = { + let mut store_guard = derived.try_write().unwrap(); + store_guard.set_pool_depths(PoolDepths::new(), vec![], 0, true); + store_guard + .spot_prices() + .unwrap() + .clone() + }; + + let mut graph = PetgraphStableDiGraphManager::new(); + graph.initialize_graph(&market_read(&market).component_topology()); + + let computation = computation_for(ð.address); + let paths = computation + .discover_paths(&graph, &spot_prices, &PoolDepths::new()) + .unwrap(); + + assert_eq!(paths[&usdc.address][0].depth, 0.0, "unmeasured depth reads as zero"); + } + // ==================== compute_spread_and_mid_price tests ==================== #[tokio::test] @@ -1013,7 +1282,40 @@ mod tests { } #[tokio::test] - async fn test_compute_selects_best_path_by_spread() { + async fn test_compute_prefers_the_deeper_of_two_pools() { + // Two quotes cannot outvote each other, so the deeper path decides. The real pool charges a + // fee and the shallow one does not, so on spread alone the shallow one — quoting a rate + // 2000x off — would have won. + let eth = token(0, "ETH"); + let usdc = token(1, "USDC"); + + let (market, derived) = setup_test_env(vec![ + ("deep", ð, &usdc, MockProtocolSim::new(2000.0).with_fee(0.01)), + ("shallow", ð, &usdc, MockProtocolSim::new(1.0)), + ]) + .await; + set_depths(&derived, vec![("shallow", ð.address, &usdc.address, BELOW_PROBE)]); + + let computation = computation_for(ð.address); + let prices = computation + .compute(&market, &derived, &ChangedComponents::default()) + .await + .unwrap() + .data; + + let usdc_price = prices + .get(&usdc.address) + .expect("USDC should have price"); + let ratio = + usdc_price.numerator.to_f64().unwrap() / usdc_price.denominator.to_f64().unwrap(); + assert!( + (1900.0..2100.0).contains(&ratio), + "price should come from the deep pool (~2000), got {ratio}" + ); + } + + #[tokio::test] + async fn test_compute_prices_a_forked_token_from_the_median_path() { // Diamond topology: two paths to C // // A (10% fee on eth_a) @@ -1106,19 +1408,18 @@ mod tests { "B mid_price should be 43377/14440 = {expected_b}, got {b_ratio}" ); - // C: Path via B selected (lower spread) - // buy_out = 1e18 * 3 * 0.95 * 2 = 5.7e18 = (57/10)e18 - // sell_out = 5.7e18 / 2 / 3 * 0.95 = 0.9025e18 = (361/400)e18 - // buy_price = 57/10, sell_price = (57/10)/(361/400) = 2280/361 - // mid_price = (57/10 + 2280/361) / 2 = (20577 + 22800) / 7220 = 43377/7220 + // C has two paths, so the median of two takes the upper middle — the one via A. + // buy_out = 1e18 * 2 * 0.9 * 5 = 9e18, so buy_price = 9 + // sell_out = 9e18 / 5 / 2 * 0.95... = (81/100)e18, so sell_price = 9 / (81/100) = 100/9 + // mid_price = (9 + 100/9) / 2 = 181/18 let c_price = prices .get(&c.address) .expect("C should have price"); let c_ratio = c_price.numerator.to_f64().unwrap() / c_price.denominator.to_f64().unwrap(); - let expected_c = 43377.0 / 7220.0; + let expected_c = 181.0 / 18.0; assert!( (c_ratio - expected_c).abs() < 1e-10, - "C mid_price should be 43377/7220 = {expected_c} (via B), got {c_ratio}" + "C mid_price should be 181/18 = {expected_c} (via A), got {c_ratio}" ); } @@ -1332,10 +1633,17 @@ mod tests { .compute(&market, &derived, &changed) .await .unwrap(); - derived - .try_write() - .unwrap() - .set_spot_prices(spot_output.data, vec![], 0, true); + let mut pool_depths = PoolDepths::new(); + pool_depths.insert( + ("pool".to_string(), eth.address.clone(), usdc.address.clone()), + BigUint::from(DEEP), + ); + { + let mut store_guard = derived.try_write().unwrap(); + store_guard.set_spot_prices(spot_output.data, vec![], 0, true); + // Present so the missing gas price is the only dependency left to fail on. + store_guard.set_pool_depths(pool_depths, vec![], 0, true); + } let computation = computation_for(ð.address); let result = computation diff --git a/fynd-core/src/derived/manager.rs b/fynd-core/src/derived/manager.rs index ae8191273..1726dab86 100644 --- a/fynd-core/src/derived/manager.rs +++ b/fynd-core/src/derived/manager.rs @@ -208,12 +208,12 @@ impl ComputationManager { ) -> Result<(Self, broadcast::Receiver), ComputationError> { let (mut manager, event_rx) = Self::empty(market_data); manager.register(SpotPriceComputation::new())?; + manager.register(PoolDepthComputation::new(config.depth_slippage_threshold)?)?; manager.register( TokenGasPriceComputation::default() .with_max_hops(config.max_hop) .with_gas_token(config.gas_token), )?; - manager.register(PoolDepthComputation::new(config.depth_slippage_threshold)?)?; Ok((manager, event_rx)) } @@ -1296,9 +1296,10 @@ mod tests { } #[tokio::test] - async fn default_computations_cascade_failure_in_registration_order() { + async fn default_computations_cascade_failure_in_dependency_order() { // Real fynd flow: a full recompute with no sim state makes spot prices fail - // outright, cascading ComputationFailed to every dependent in registration order. + // outright, cascading ComputationFailed to every dependent. Token prices require both + // spot prices and depths, so they come last. let (manager, _event_rx) = ComputationManager::new( ComputationManagerConfig::new(), market_with_component_no_sim_state(), @@ -1312,8 +1313,8 @@ mod tests { vec![ ("new_block", ""), ("failed", "spot_prices"), - ("failed", "token_prices"), ("failed", "pool_depths"), + ("failed", "token_prices"), ] ); }