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"), ] ); }