1use std::{
32 cmp::{Ordering, Reverse},
33 collections::{HashMap, HashSet},
34 time::{Duration, Instant},
35};
36
37use num_bigint::{BigInt, BigUint};
38use num_traits::Zero;
39use petgraph::{graph::NodeIndex, prelude::EdgeRef};
40use tracing::{debug, instrument};
41use tycho_simulation::{
42 tycho_common::simulation::protocol_sim::ProtocolSim, tycho_core::models::Address,
43};
44
45use super::{
46 most_liquid::DepthAndPrice,
47 split_primitives::{build_split_route, HopDescriptor, PathAllocation, SimulatedHop},
48 Algorithm, AlgorithmConfig, MostLiquidAlgorithm, NoPathReason,
49};
50use crate::{
51 derived::{computation::ComputationRequirements, types::TokenGasPrices, SharedDerivedDataRef},
52 feed::market_data::{MarketData, MarketDataView, MarketState, StateLabel},
53 graph::{petgraph::StableDiGraph, EdgeData, Path, PetgraphStableDiGraphManager},
54 types::{ComponentId, Order, Route, RouteResult},
55 AlgorithmError,
56};
57
58const DEFAULT_MAX_CANDIDATES: usize = 5000;
60const BOUNDED_DISCOVERY_CANDIDATES: usize = 128;
63const DEFAULT_MAX_PATHS: usize = 4;
65const COARSE_CHUNKS: usize = 20;
67const FINE_CHUNKS: usize = 256;
69const SHARED_FULL_PATHS: usize = 8;
71const SHARED_MARGIN_PROBE_PATHS: usize = 32;
73const SHARED_MARGIN_PATHS: usize = 8;
75const SHARED_MAX_CANDIDATES: usize = 12;
77const CANDIDATE_STATES_PER_NODE: usize = 4;
79const CANDIDATE_EDGES_PER_STATE: usize = 16;
81const CANDIDATE_DIRECT_EDGES_PER_TOKEN: usize = 4;
83const CANDIDATE_CONNECTOR_EDGES_PER_TOKEN: usize = 2;
85const DERIVED_ANCHOR_COUNT: usize = 16;
88const EXCHANGE_DELTA_FLOOR: usize = 64;
91const EXCHANGE_MAX_SIMS: usize = 400;
93
94struct SplitCandidate {
96 route: Route,
97 gross: BigUint,
98 gas: BigUint,
99}
100
101impl SplitCandidate {
102 fn net(
104 &self,
105 gas_price: &BigUint,
106 token_prices: Option<&TokenGasPrices>,
107 token_out: &Address,
108 ) -> BigInt {
109 let cost =
110 WaterFillAlgorithm::gas_cost_in_token(&self.gas, gas_price, token_prices, token_out);
111 match cost {
112 Some(c) => BigInt::from(self.gross.clone()) - BigInt::from(c),
113 None => BigInt::from(self.gross.clone()),
114 }
115 }
116}
117
118pub struct WaterFillAlgorithm {
121 min_hops: usize,
122 max_hops: usize,
123 timeout: Duration,
124 max_candidates: usize,
125 max_paths: usize,
126 connector_tokens: Option<HashSet<Address>>,
127}
128
129struct ExchangeMove {
133 donor: usize,
134 recipient: usize,
135 donor_net: BigInt,
136 recip_net: BigInt,
137 gain: BigInt,
138}
139
140struct StepResult {
142 amount_out: BigUint,
143 gas: BigUint,
144 new_states: Vec<(ComponentId, Box<dyn ProtocolSim>)>,
145}
146
147struct SplitContext<'a, 'g> {
151 ordered: &'a [Path<'g, DepthAndPrice>],
152 market: &'a MarketState,
153 gas_price: &'a BigUint,
154 token_prices: Option<&'a TokenGasPrices>,
155 order: &'a Order,
156 start: Instant,
157}
158
159struct SetupResult<'a> {
163 ordered: Vec<Path<'a, DepthAndPrice>>,
164 market: MarketState,
165 gas_price: BigUint,
166 best_single: Option<RouteResult>,
167 token_prices: Option<TokenGasPrices>,
168}
169
170impl WaterFillAlgorithm {
171 pub(crate) fn with_config(config: AlgorithmConfig) -> Result<Self, AlgorithmError> {
173 Ok(Self {
174 min_hops: config.min_hops(),
175 max_hops: config.max_hops(),
176 timeout: config.timeout(),
177 max_candidates: config
178 .max_routes()
179 .unwrap_or(DEFAULT_MAX_CANDIDATES)
180 .max(DEFAULT_MAX_PATHS),
181 max_paths: DEFAULT_MAX_PATHS,
182 connector_tokens: config.connector_tokens().cloned(),
183 })
184 }
185
186 fn simulate_step(
190 path: &Path<DepthAndPrice>,
191 market: &MarketState,
192 overlay: &HashMap<ComponentId, Box<dyn ProtocolSim>>,
193 amount: BigUint,
194 ) -> Option<StepResult> {
195 let mut current = amount;
196 let mut total_gas = BigUint::zero();
197 let path_reuses_pool = {
201 let mut seen: HashSet<&ComponentId> = HashSet::with_capacity(path.len());
202 !path
203 .edge_iter()
204 .iter()
205 .all(|e| seen.insert(&e.component_id))
206 };
207 let mut intra_path_states: HashMap<ComponentId, Box<dyn ProtocolSim>> = HashMap::new();
208 let mut new_states: Vec<(ComponentId, Box<dyn ProtocolSim>)> =
209 Vec::with_capacity(path.len());
210
211 for (address_in, edge, address_out) in path.iter() {
212 let token_in = market.get_token(address_in)?;
213 let token_out = market.get_token(address_out)?;
214 let component_id = &edge.component_id;
215 let state = intra_path_states
216 .get(component_id)
217 .map(Box::as_ref)
218 .or_else(|| {
219 overlay
220 .get(component_id)
221 .map(Box::as_ref)
222 })
223 .or_else(|| market.get_simulation_state(component_id))?;
224 let result = state
225 .get_amount_out(current.clone(), token_in, token_out)
226 .ok()?;
227 total_gas += &result.gas;
228 if path_reuses_pool {
229 intra_path_states.insert(component_id.clone(), result.new_state.clone_box());
230 }
231 new_states.push((component_id.clone(), result.new_state));
232 current = result.amount;
233 }
234 Some(StepResult { amount_out: current, gas: total_gas, new_states })
235 }
236
237 fn gas_cost_in_token(
239 total_gas: &BigUint,
240 gas_price_wei: &BigUint,
241 token_prices: Option<&TokenGasPrices>,
242 token_out: &Address,
243 ) -> Option<BigUint> {
244 let price = token_prices?.get(token_out)?;
245 if price.denominator.is_zero() {
246 return None;
247 }
248 Some(total_gas * gas_price_wei * &price.numerator / &price.denominator)
249 }
250
251 fn activation_cost(ctx: &SplitContext, gas: &BigUint) -> BigInt {
255 Self::gas_cost_in_token(gas, ctx.gas_price, ctx.token_prices, ctx.order.token_out())
256 .map(BigInt::from)
257 .unwrap_or_else(BigInt::zero)
258 }
259
260 fn select_disjoint(ranked: &[Path<DepthAndPrice>], max_paths: usize) -> Vec<usize> {
267 let mut visited_pools: HashSet<&ComponentId> = HashSet::new();
268 let mut selected = Vec::new();
269 for (idx, path) in ranked.iter().enumerate() {
270 let path_pools: Vec<&ComponentId> = path
271 .edge_iter()
272 .iter()
273 .map(|e| &e.component_id)
274 .collect();
275 if path_pools
276 .iter()
277 .any(|c| visited_pools.contains(*c))
278 {
279 continue;
280 }
281 for c in path_pools {
282 visited_pools.insert(c);
283 }
284 selected.push(idx);
285 if selected.len() >= max_paths {
286 break;
287 }
288 }
289 selected
290 }
291
292 #[instrument(level = "debug", skip_all)]
295 async fn setup<'a>(
296 &self,
297 graph: &'a StableDiGraph<DepthAndPrice>,
298 market: MarketData,
299 label: Option<StateLabel>,
300 derived: Option<SharedDerivedDataRef>,
301 order: &Order,
302 start: Instant,
303 ) -> Result<SetupResult<'a>, AlgorithmError> {
304 let token_prices = if let Some(ref derived) = derived {
305 derived
306 .read()
307 .await
308 .token_prices()
309 .cloned()
310 } else {
311 None
312 };
313
314 let all_paths = MostLiquidAlgorithm::find_paths(
315 graph,
316 order.token_in(),
317 order.token_out(),
318 self.min_hops,
319 self.max_hops,
320 self.connector_tokens.as_ref(),
321 )?;
322 if all_paths.is_empty() {
323 return Err(AlgorithmError::NoPath {
324 from: order.token_in().clone(),
325 to: order.token_out().clone(),
326 reason: NoPathReason::NoGraphPath,
327 });
328 }
329
330 let mut scored: Vec<(Path<DepthAndPrice>, f64)> = all_paths
331 .into_iter()
332 .map(|p| {
333 let s = MostLiquidAlgorithm::try_score_path(&p).unwrap_or(f64::MIN);
334 (p, s)
335 })
336 .collect();
337 scored.sort_by(|(_, a), (_, b)| {
338 b.partial_cmp(a)
339 .unwrap_or(std::cmp::Ordering::Equal)
340 });
341 scored.truncate(self.max_candidates);
342 let mut paths: Vec<Path<DepthAndPrice>> = scored
343 .into_iter()
344 .map(|(p, _)| p)
345 .collect();
346
347 let timeout_ms = self.timeout.as_millis() as u64;
348 let market = {
349 let view = match label.as_ref() {
350 Some(l) => market
351 .read_labeled(l)
352 .await
353 .map_err(|e| AlgorithmError::Other(e.to_string()))?,
354 None => market.read().await,
355 };
356 if view.gas_price().is_none() {
357 return Err(AlgorithmError::DataNotFound { kind: "gas price", id: None });
358 }
359 let anchor_tokens = derive_anchor_tokens(graph);
364 let bounded = find_candidate_paths(
365 graph,
366 &view,
367 order,
368 CandidateSearchConfig {
369 min_hops: self.min_hops,
370 max_hops: self.max_hops,
371 max_candidates: BOUNDED_DISCOVERY_CANDIDATES,
372 connector_tokens: self.connector_tokens.as_ref(),
373 anchor_tokens: &anchor_tokens,
374 source_token: order.token_in(),
375 start: &start,
376 timeout_ms,
377 },
378 );
379 match bounded {
380 Ok((bounded_paths, _)) => {
381 let mut keys: HashSet<Vec<ComponentId>> = paths.iter().map(path_key).collect();
382 let mut union = Vec::with_capacity(bounded_paths.len() + paths.len());
383 for path in bounded_paths {
384 if keys.insert(path_key(&path)) {
385 union.push(path);
386 }
387 }
388 union.append(&mut paths);
389 paths = union;
390 }
391 Err(e) => {
395 debug!(error = %e, "water-fill bounded discovery failed; using exhaustive candidates only")
396 }
397 }
398 let component_ids: HashSet<ComponentId> = paths
399 .iter()
400 .flat_map(|p| {
401 p.edge_iter()
402 .iter()
403 .map(|e| e.component_id.clone())
404 })
405 .collect();
406 let subset = view.extract_subset_with_overlay(&component_ids);
407 drop(view);
408 subset
409 };
410 let gas_price = market
411 .gas_price()
412 .ok_or(AlgorithmError::DataNotFound { kind: "gas price", id: None })?
413 .effective_gas_price()
414 .clone();
415
416 let amount_in = order.amount().clone();
417 let mut best_single: Option<RouteResult> = None;
418 let mut full_outputs: Vec<(usize, BigUint)> = Vec::new();
419
420 for (idx, path) in paths.iter().enumerate() {
421 if start.elapsed().as_millis() as u64 > timeout_ms {
422 break;
423 }
424 match MostLiquidAlgorithm::simulate_path(
425 path,
426 &market,
427 token_prices.as_ref(),
428 amount_in.clone(),
429 ) {
430 Ok(result) => {
431 let gross = result
432 .route()
433 .swaps()
434 .last()
435 .map(|s| s.amount_out().clone())
436 .unwrap_or_else(BigUint::zero);
437 full_outputs.push((idx, gross));
438 if best_single
439 .as_ref()
440 .map(|b| result.net_amount_out() > b.net_amount_out())
441 .unwrap_or(true)
442 {
443 best_single = Some(result);
444 }
445 }
446 Err(_) => full_outputs.push((idx, BigUint::zero())),
451 }
452 }
453
454 full_outputs.sort_by(|(_, a), (_, b)| b.cmp(a));
458 let ordered: Vec<Path<DepthAndPrice>> = full_outputs
459 .into_iter()
460 .map(|(idx, _)| paths[idx].clone())
461 .collect();
462
463 debug!(
464 candidate_paths = ordered.len(),
465 elapsed_ms = start.elapsed().as_millis(),
466 "water-fill discovery + full-amount ranking"
467 );
468 Ok(SetupResult { ordered, market, gas_price, best_single, token_prices })
469 }
470}
471
472impl Algorithm for WaterFillAlgorithm {
473 type GraphType = StableDiGraph<DepthAndPrice>;
474 type GraphManager = PetgraphStableDiGraphManager<DepthAndPrice>;
475
476 fn name(&self) -> &str {
477 "water_fill"
478 }
479
480 #[instrument(level = "debug", skip_all, fields(order_id = %order.id()))]
481 async fn find_best_route(
482 &self,
483 graph: &Self::GraphType,
484 market: MarketData,
485 label: Option<StateLabel>,
486 derived: Option<SharedDerivedDataRef>,
487 order: &Order,
488 ) -> Result<RouteResult, AlgorithmError> {
489 let start = Instant::now();
490 if !order.is_sell() {
491 return Err(AlgorithmError::ExactOutNotSupported);
492 }
493
494 let SetupResult { ordered, market, gas_price, best_single, token_prices } = self
495 .setup(graph, market, label, derived, order, start)
496 .await?;
497 let token_out = order.token_out();
498 let ctx = SplitContext {
499 ordered: &ordered,
500 market: &market,
501 gas_price: &gas_price,
502 token_prices: token_prices.as_ref(),
503 order,
504 start,
505 };
506
507 let mut candidates: Vec<SplitCandidate> = Vec::new();
509 let disjoint = Self::select_disjoint(ctx.ordered, self.max_paths);
514 let coarse = (disjoint.len() >= 2)
515 .then(|| self.disjoint_waterfill(&ctx, &disjoint, COARSE_CHUNKS, true))
516 .flatten();
517 if let Some(coarse) = coarse.as_deref() {
518 if let Some(c) = self.build_disjoint_legs(&ctx, &disjoint, coarse) {
520 candidates.push(c);
521 }
522 if let Some(c) = self.disjoint_refine(&ctx, &disjoint, coarse, FINE_CHUNKS) {
525 candidates.push(c);
526 }
527 }
528 if let Some(c) = self.fillspill_alloc(&ctx, FINE_CHUNKS) {
531 candidates.push(c);
532 }
533
534 let candidate_count = candidates.len();
537 let baseline_net = best_single
538 .as_ref()
539 .map(|b| b.net_amount_out().clone());
540 let mut best: Option<(BigInt, SplitCandidate)> = None;
541 for cand in candidates {
542 let net = cand.net(&gas_price, token_prices.as_ref(), token_out);
543 let beats = match (&best, &baseline_net) {
544 (Some((current, _)), _) => net > *current,
545 (None, Some(base)) => net > *base,
546 (None, None) => true,
547 };
548 if beats {
549 best = Some((net, cand));
550 }
551 }
552 let split_won = best.is_some();
553 debug!(
554 candidate_count,
555 split_won,
556 elapsed_ms = start.elapsed().as_millis(),
557 "water-fill selected {}",
558 if split_won { "split candidate" } else { "single path" }
559 );
560 match best {
561 Some((net, cand)) => Ok(RouteResult::new(cand.route, net, gas_price)),
562 None => best_single.ok_or(AlgorithmError::InsufficientLiquidity),
564 }
565 }
566
567 fn computation_requirements(&self) -> ComputationRequirements {
568 ComputationRequirements::none()
569 .allow_stale("token_prices")
570 .expect("Conflicting Computation Requirements")
571 }
572
573 fn timeout(&self) -> Duration {
574 self.timeout
575 }
576}
577
578impl WaterFillAlgorithm {
579 fn disjoint_refine(
585 &self,
586 ctx: &SplitContext,
587 disjoint: &[usize],
588 coarse: &[BigUint],
589 fine_chunks: usize,
590 ) -> Option<SplitCandidate> {
591 let active: Vec<usize> = disjoint
593 .iter()
594 .copied()
595 .zip(coarse.iter())
596 .filter(|(_, amt)| !amt.is_zero())
597 .map(|(idx, _)| idx)
598 .collect();
599 if active.is_empty() {
600 return None;
601 }
602
603 let fine = self.disjoint_waterfill(ctx, &active, fine_chunks, false)?;
605
606 let refined = self.disjoint_exchange(ctx, &active, fine_chunks, fine);
610 self.build_disjoint_legs(ctx, &active, &refined)
611 }
612
613 fn path_net(
619 ctx: &SplitContext,
620 path: &Path<DepthAndPrice>,
621 amount: &BigUint,
622 ) -> Option<BigInt> {
623 if amount.is_zero() {
624 return Some(BigInt::zero());
625 }
626 let empty: HashMap<ComponentId, Box<dyn ProtocolSim>> = HashMap::new();
627 let step = Self::simulate_step(path, ctx.market, &empty, amount.clone())?;
628 let activation = Self::activation_cost(ctx, &step.gas);
629 Some(BigInt::from(step.amount_out) - activation)
630 }
631
632 fn disjoint_exchange(
641 &self,
642 ctx: &SplitContext,
643 active: &[usize],
644 fine_chunks: usize,
645 alloc: Vec<BigUint>,
646 ) -> Vec<BigUint> {
647 let path_count = active.len();
648 if path_count < 2 {
649 return alloc;
650 }
651 let amount_in = ctx.order.amount().clone();
652 let fine_chunks = fine_chunks.max(1);
653 let mut delta = &amount_in / fine_chunks;
654 let min_delta = &amount_in / (fine_chunks * EXCHANGE_DELTA_FLOOR);
655 if delta.is_zero() {
656 return alloc;
657 }
658 let timeout_ms = self.timeout.as_millis() as u64;
659
660 let mut cum = alloc;
661 let mut net_cache: Vec<BigInt> = Vec::with_capacity(path_count);
664 for (i, &path_idx) in active.iter().enumerate() {
665 let Some(net) = Self::path_net(ctx, &ctx.ordered[path_idx], &cum[i]) else {
666 return cum;
668 };
669 net_cache.push(net);
670 }
671
672 let mut sims = 0usize;
673 while delta >= min_delta && !delta.is_zero() {
674 if ctx.start.elapsed().as_millis() as u64 > timeout_ms || sims >= EXCHANGE_MAX_SIMS {
675 break;
676 }
677
678 let mut best: Option<ExchangeMove> = None;
679 for donor in 0..path_count {
680 if sims >= EXCHANGE_MAX_SIMS {
681 break;
682 }
683 if cum[donor] < delta {
684 continue;
685 }
686 let donor_amt = &cum[donor] - δ
687 let Some(donor_net) = Self::path_net(ctx, &ctx.ordered[active[donor]], &donor_amt)
688 else {
689 continue;
690 };
691 sims += 1;
692 for recipient in 0..path_count {
693 if recipient == donor || sims >= EXCHANGE_MAX_SIMS {
694 continue;
695 }
696 let recip_amt = &cum[recipient] + δ
697 let Some(recip_net) =
698 Self::path_net(ctx, &ctx.ordered[active[recipient]], &recip_amt)
699 else {
700 continue;
701 };
702 sims += 1;
703 let before = &net_cache[donor] + &net_cache[recipient];
704 let after = &donor_net + &recip_net;
705 if after <= before {
706 continue;
707 }
708 let gain = after - before;
709 if best
710 .as_ref()
711 .map(|m| gain > m.gain)
712 .unwrap_or(true)
713 {
714 best = Some(ExchangeMove {
715 donor,
716 recipient,
717 donor_net: donor_net.clone(),
718 recip_net,
719 gain,
720 });
721 }
722 }
723 }
724
725 let Some(mv) = best else {
726 delta = &delta / 2usize;
727 continue;
728 };
729 cum[mv.donor] = &cum[mv.donor] - δ
730 cum[mv.recipient] = &cum[mv.recipient] + δ
731 net_cache[mv.donor] = mv.donor_net;
732 net_cache[mv.recipient] = mv.recip_net;
733 }
734 cum
735 }
736
737 fn allocation_commit(
740 path: &Path<DepthAndPrice>,
741 market: &MarketState,
742 overrides: &mut HashMap<ComponentId, Box<dyn ProtocolSim>>,
743 amount: BigUint,
744 flow_fraction: f64,
745 ) -> Option<PathAllocation> {
746 let amount_in = amount;
747 let mut current = amount_in.clone();
748 let mut hops = Vec::with_capacity(path.len());
749
750 for (address_in, edge, address_out) in path.iter() {
751 let token_in = market.get_token(address_in)?;
752 let token_out = market.get_token(address_out)?;
753 let component_id = &edge.component_id;
754 let state = overrides
755 .get(component_id)
756 .map(Box::as_ref)
757 .or_else(|| market.get_simulation_state(component_id))?;
758 let result = state
759 .get_amount_out(current.clone(), token_in, token_out)
760 .ok()?;
761 hops.push(SimulatedHop {
762 descriptor: HopDescriptor::new(
763 component_id.clone(),
764 token_in.clone(),
765 token_out.clone(),
766 ),
767 amount_out: result.amount.clone(),
768 gas: result.gas.clone(),
769 });
770 overrides.insert(component_id.clone(), result.new_state);
771 current = result.amount;
772 }
773
774 Some(PathAllocation {
775 hops,
776 flow_fraction,
777 amount_in,
778 amount_out: current,
779 marginal_price_product: 0.0,
780 })
781 }
782
783 fn candidate_from_allocations(
787 ctx: &SplitContext,
788 allocations: &[PathAllocation],
789 ) -> Option<SplitCandidate> {
790 let route = build_split_route(allocations, ctx.market, ctx.order).ok()?;
791 let token_out = ctx.order.token_out();
792 let gross = route
793 .swaps()
794 .iter()
795 .filter(|s| s.token_out() == token_out)
796 .fold(BigUint::zero(), |acc, s| acc + s.amount_out());
797 if gross.is_zero() {
798 return None;
799 }
800 let gas = route.total_gas();
801 Some(SplitCandidate { route, gross, gas })
802 }
803
804 fn build_disjoint_legs(
808 &self,
809 ctx: &SplitContext,
810 subset: &[usize],
811 alloc: &[BigUint],
812 ) -> Option<SplitCandidate> {
813 let amount_in = ctx.order.amount().clone();
814 let mut allocations = Vec::new();
815 for (i, &path_idx) in subset.iter().enumerate() {
816 if alloc[i].is_zero() {
817 continue;
818 }
819 let mut overrides: HashMap<ComponentId, Box<dyn ProtocolSim>> = HashMap::new();
822 let allocation = Self::allocation_commit(
823 &ctx.ordered[path_idx],
824 ctx.market,
825 &mut overrides,
826 alloc[i].clone(),
827 ratio(&alloc[i], &amount_in),
828 )?;
829 allocations.push(allocation);
830 }
831 if allocations.is_empty() {
832 return None;
833 }
834 Self::candidate_from_allocations(ctx, &allocations)
835 }
836
837 fn disjoint_waterfill(
841 &self,
842 ctx: &SplitContext,
843 subset: &[usize],
844 num_chunks: usize,
845 gate: bool,
846 ) -> Option<Vec<BigUint>> {
847 let amount_in = ctx.order.amount().clone();
848 let num_chunks = num_chunks.max(1);
849 let base_chunk = &amount_in / num_chunks;
850 if base_chunk.is_zero() {
851 return None;
852 }
853 let remainder = &amount_in - &base_chunk * num_chunks;
854 let timeout_ms = self.timeout.as_millis() as u64;
855 let path_count = subset.len();
856
857 let mut committed: Vec<HashMap<ComponentId, Box<dyn ProtocolSim>>> = (0..path_count)
858 .map(|_| HashMap::new())
859 .collect();
860 let mut cum_in: Vec<BigUint> = vec![BigUint::zero(); path_count];
861 let mut activated: Vec<bool> = vec![!gate; path_count];
862
863 for chunk_idx in 0..num_chunks {
864 if ctx.start.elapsed().as_millis() as u64 > timeout_ms {
865 break;
866 }
867 let chunk = if chunk_idx == 0 { &base_chunk + &remainder } else { base_chunk.clone() };
868
869 let mut best: Option<(usize, BigInt, StepResult)> = None;
870 for (i, &path_idx) in subset.iter().enumerate() {
871 let Some(step) = Self::simulate_step(
872 &ctx.ordered[path_idx],
873 ctx.market,
874 &committed[i],
875 chunk.clone(),
876 ) else {
877 continue;
878 };
879 let gross_marginal = BigInt::from(step.amount_out.clone());
880 let net_marginal = if activated[i] {
881 gross_marginal
882 } else {
883 let activation = Self::activation_cost(ctx, &step.gas);
884 gross_marginal - activation
885 };
886 if best
887 .as_ref()
888 .map(|(_, m, _)| &net_marginal > m)
889 .unwrap_or(true)
890 {
891 best = Some((i, net_marginal, step));
892 }
893 }
894
895 let Some((best_i, _, step)) = best else {
896 break;
897 };
898 for (id, state) in step.new_states {
899 committed[best_i].insert(id, state);
900 }
901 cum_in[best_i] += &chunk;
902 activated[best_i] = true;
903 }
904 Some(cum_in)
905 }
906
907 fn select_shared_candidates(&self, ctx: &SplitContext) -> Vec<usize> {
911 let mut candidates: Vec<usize> = (0..ctx.ordered.len().min(SHARED_FULL_PATHS)).collect();
912 let first_chunk = ctx.order.amount() / COARSE_CHUNKS;
913 if first_chunk.is_zero() {
914 return candidates;
915 }
916 let timeout_ms = self.timeout.as_millis() as u64;
917 let empty: HashMap<ComponentId, Box<dyn ProtocolSim>> = HashMap::new();
918 let mut marginal: Vec<(usize, BigInt)> = Vec::new();
919 for (idx, path) in ctx
920 .ordered
921 .iter()
922 .enumerate()
923 .take(SHARED_MARGIN_PROBE_PATHS)
924 {
925 if ctx.start.elapsed().as_millis() as u64 > timeout_ms {
926 break;
927 }
928 let Some(step) = Self::simulate_step(path, ctx.market, &empty, first_chunk.clone())
929 else {
930 continue;
931 };
932 let activation = Self::activation_cost(ctx, &step.gas);
933 marginal.push((idx, BigInt::from(step.amount_out) - activation));
934 }
935 marginal.sort_by(|(_, a), (_, b)| b.cmp(a));
936 for (idx, net) in marginal
937 .into_iter()
938 .take(SHARED_MARGIN_PATHS)
939 {
940 if net <= BigInt::zero() {
941 continue;
942 }
943 if !candidates.contains(&idx) {
944 candidates.push(idx);
945 }
946 if candidates.len() >= SHARED_MAX_CANDIDATES {
947 break;
948 }
949 }
950 candidates
951 }
952
953 fn fillspill_alloc(&self, ctx: &SplitContext, fine_chunks: usize) -> Option<SplitCandidate> {
955 let candidates = self.select_shared_candidates(ctx);
956 if candidates.len() < 2 {
957 return None;
958 }
959
960 let (coarse_counts, _) = self.fillspill_waterfill(ctx, &candidates, COARSE_CHUNKS, true)?;
962 let active: Vec<usize> = candidates
963 .iter()
964 .copied()
965 .zip(coarse_counts.iter())
966 .filter(|(_, count)| **count > 0)
967 .map(|(idx, _)| idx)
968 .collect();
969 if active.len() < 2 {
970 return None;
971 }
972
973 let (_, schedule) = self.fillspill_waterfill(ctx, &active, fine_chunks, false)?;
975 if schedule.is_empty() {
976 return None;
977 }
978
979 self.build_fillspill_route(ctx, &active, &schedule)
980 }
981
982 #[allow(clippy::type_complexity)]
985 fn fillspill_waterfill(
986 &self,
987 ctx: &SplitContext,
988 subset: &[usize],
989 num_chunks: usize,
990 gate: bool,
991 ) -> Option<(Vec<usize>, Vec<(usize, BigUint)>)> {
992 let amount_in = ctx.order.amount().clone();
993 let num_chunks = num_chunks.max(1);
994 let base_chunk = &amount_in / num_chunks;
995 if base_chunk.is_zero() {
996 return None;
997 }
998 let remainder = &amount_in - &base_chunk * num_chunks;
999 let timeout_ms = self.timeout.as_millis() as u64;
1000
1001 let mut overlay: HashMap<ComponentId, Box<dyn ProtocolSim>> = HashMap::new();
1002 let mut activated: Vec<bool> = vec![!gate; subset.len()];
1003 let mut active_count = if gate { 0 } else { subset.len() };
1004 let mut counts: Vec<usize> = vec![0; subset.len()];
1005 let mut schedule: Vec<(usize, BigUint)> = Vec::with_capacity(num_chunks);
1006
1007 for chunk_idx in 0..num_chunks {
1008 if ctx.start.elapsed().as_millis() as u64 > timeout_ms {
1009 break;
1010 }
1011 let chunk = if chunk_idx == 0 { &base_chunk + &remainder } else { base_chunk.clone() };
1012
1013 let mut best: Option<(usize, BigInt, StepResult)> = None;
1014 for (i, &path_idx) in subset.iter().enumerate() {
1015 if !activated[i] && active_count >= self.max_paths {
1016 continue;
1017 }
1018 let Some(step) = Self::simulate_step(
1019 &ctx.ordered[path_idx],
1020 ctx.market,
1021 &overlay,
1022 chunk.clone(),
1023 ) else {
1024 continue;
1025 };
1026 let gross_marginal = BigInt::from(step.amount_out.clone());
1027 let net_marginal = if activated[i] {
1028 gross_marginal
1029 } else {
1030 let activation = Self::activation_cost(ctx, &step.gas);
1031 gross_marginal - activation
1032 };
1033 if best
1034 .as_ref()
1035 .map(|(_, m, _)| &net_marginal > m)
1036 .unwrap_or(true)
1037 {
1038 best = Some((i, net_marginal, step));
1039 }
1040 }
1041
1042 let Some((best_i, _, step)) = best else {
1043 break;
1044 };
1045 for (id, state) in step.new_states {
1046 overlay.insert(id, state);
1047 }
1048 if !activated[best_i] {
1049 activated[best_i] = true;
1050 active_count += 1;
1051 }
1052 counts[best_i] += 1;
1053 schedule.push((best_i, chunk));
1054 }
1055 Some((counts, schedule))
1056 }
1057
1058 fn build_fillspill_route(
1062 &self,
1063 ctx: &SplitContext,
1064 active: &[usize],
1065 schedule: &[(usize, BigUint)],
1066 ) -> Option<SplitCandidate> {
1067 let amount_in = ctx.order.amount().clone();
1068 let mut cand_in: Vec<BigUint> = vec![BigUint::zero(); active.len()];
1069 for (i, chunk) in schedule {
1070 cand_in[*i] += chunk;
1071 }
1072 let mut execution_order: Vec<usize> = (0..active.len())
1073 .filter(|&i| !cand_in[i].is_zero())
1074 .collect();
1075 if execution_order.len() < 2 {
1076 return None;
1077 }
1078 execution_order.sort_by(|&a, &b| cand_in[b].cmp(&cand_in[a]));
1079
1080 let mut overrides: HashMap<ComponentId, Box<dyn ProtocolSim>> = HashMap::new();
1081 let mut allocations = Vec::new();
1082 for i in execution_order {
1083 let allocation = Self::allocation_commit(
1084 &ctx.ordered[active[i]],
1085 ctx.market,
1086 &mut overrides,
1087 cand_in[i].clone(),
1088 ratio(&cand_in[i], &amount_in),
1089 )?;
1090 allocations.push(allocation);
1091 }
1092 Self::candidate_from_allocations(ctx, &allocations)
1093 }
1094}
1095
1096fn path_key<W>(path: &Path<'_, W>) -> Vec<ComponentId> {
1098 path.edge_iter()
1099 .iter()
1100 .map(|e| e.component_id.clone())
1101 .collect()
1102}
1103
1104fn ratio(numerator: &BigUint, denominator: &BigUint) -> f64 {
1106 use num_traits::ToPrimitive;
1107 let n = numerator.to_f64().unwrap_or(0.0);
1108 let d = denominator.to_f64().unwrap_or(1.0);
1109 if d == 0.0 {
1110 0.0
1111 } else {
1112 n / d
1113 }
1114}
1115
1116type RankedPathScores = Vec<(usize, BigInt)>;
1127type CandidatePathSet<'a, W = ()> = (Vec<Path<'a, W>>, RankedPathScores);
1128
1129#[derive(Clone)]
1130struct CandidatePathState<'a, W> {
1131 node: NodeIndex,
1132 path: Path<'a, W>,
1133 amount_out: BigUint,
1134}
1135
1136struct ScoredEdge<'a, W> {
1137 target: NodeIndex,
1138 edge: &'a EdgeData<W>,
1139 amount_out: BigUint,
1140 priority: u8,
1141}
1142
1143#[derive(Clone, Copy)]
1145struct CandidateSearchConfig<'a> {
1146 min_hops: usize,
1147 max_hops: usize,
1148 max_candidates: usize,
1149 connector_tokens: Option<&'a HashSet<Address>>,
1150 anchor_tokens: &'a HashSet<Address>,
1151 source_token: &'a Address,
1152 start: &'a Instant,
1153 timeout_ms: u64,
1154}
1155
1156fn timed_out(start: &Instant, timeout_ms: u64) -> bool {
1157 start.elapsed().as_millis() as u64 > timeout_ms
1158}
1159
1160fn find_candidate_paths<'a, W>(
1163 graph: &'a StableDiGraph<W>,
1164 market: &MarketDataView<'_>,
1165 order: &Order,
1166 cfg: CandidateSearchConfig<'_>,
1167) -> Result<CandidatePathSet<'a, W>, AlgorithmError>
1168where
1169 W: Clone,
1170{
1171 if cfg.min_hops == 0 || cfg.min_hops > cfg.max_hops {
1172 return Err(AlgorithmError::InvalidConfiguration {
1173 reason: format!(
1174 "invalid hop configuration: min_hops={} max_hops={}",
1175 cfg.min_hops, cfg.max_hops,
1176 ),
1177 });
1178 }
1179 let from_idx =
1180 find_token_node(graph, order.token_in(), NoPathReason::SourceTokenNotInGraph, order)?;
1181 let to_idx =
1182 find_token_node(graph, order.token_out(), NoPathReason::DestinationTokenNotInGraph, order)?;
1183
1184 let mut found = Vec::new();
1185 let mut frontier = vec![CandidatePathState {
1186 node: from_idx,
1187 path: Path::new(),
1188 amount_out: order.amount().clone(),
1189 }];
1190
1191 for _depth in 0..cfg.max_hops {
1192 if timed_out(cfg.start, cfg.timeout_ms) || frontier.is_empty() {
1193 break;
1194 }
1195 let mut next_by_node: HashMap<NodeIndex, Vec<CandidatePathState<'a, W>>> = HashMap::new();
1196 for state in frontier {
1197 if state.node == to_idx && from_idx != to_idx {
1198 continue;
1199 }
1200 expand_candidate_state(
1201 graph,
1202 market,
1203 &cfg,
1204 to_idx,
1205 state,
1206 &mut found,
1207 &mut next_by_node,
1208 );
1209 }
1210 frontier = prune_candidate_frontier(next_by_node);
1211 }
1212
1213 rank_found_candidate_paths(found, cfg.max_candidates, order)
1214}
1215
1216fn find_token_node<W>(
1217 graph: &StableDiGraph<W>,
1218 token: &Address,
1219 reason: NoPathReason,
1220 order: &Order,
1221) -> Result<NodeIndex, AlgorithmError> {
1222 graph
1223 .node_indices()
1224 .find(|&node| &graph[node] == token)
1225 .ok_or(AlgorithmError::NoPath {
1226 from: order.token_in().clone(),
1227 to: order.token_out().clone(),
1228 reason,
1229 })
1230}
1231
1232fn expand_candidate_state<'a, W>(
1233 graph: &'a StableDiGraph<W>,
1234 market: &MarketDataView<'_>,
1235 cfg: &CandidateSearchConfig<'_>,
1236 target: NodeIndex,
1237 state: CandidatePathState<'a, W>,
1238 found: &mut Vec<(Path<'a, W>, BigUint)>,
1239 next_by_node: &mut HashMap<NodeIndex, Vec<CandidatePathState<'a, W>>>,
1240) where
1241 W: Clone,
1242{
1243 let edges = candidate_edges_for_state(graph, market, cfg, target, &state);
1244 for candidate in edges {
1245 if timed_out(cfg.start, cfg.timeout_ms) {
1246 break;
1247 }
1248 let mut path = state.path.clone();
1249 path.add_hop(&graph[state.node], candidate.edge, &graph[candidate.target]);
1250 let path_state = CandidatePathState {
1251 node: candidate.target,
1252 path: path.clone(),
1253 amount_out: candidate.amount_out,
1254 };
1255 if candidate.target == target && path.len() >= cfg.min_hops {
1256 found.push((path.clone(), path_state.amount_out.clone()));
1257 }
1258 if path.len() < cfg.max_hops {
1259 next_by_node
1260 .entry(candidate.target)
1261 .or_default()
1262 .push(path_state);
1263 }
1264 }
1265}
1266
1267fn candidate_edges_for_state<'a, W>(
1268 graph: &'a StableDiGraph<W>,
1269 market: &MarketDataView<'_>,
1270 cfg: &CandidateSearchConfig<'_>,
1271 target: NodeIndex,
1272 state: &CandidatePathState<'a, W>,
1273) -> Vec<ScoredEdge<'a, W>> {
1274 let mut preferred = score_candidate_edges(graph, market, cfg, target, state, true);
1275 if preferred.is_empty() {
1276 preferred = score_candidate_edges(graph, market, cfg, target, state, false);
1277 }
1278 select_candidate_edges(preferred, CANDIDATE_EDGES_PER_STATE)
1279}
1280
1281fn score_candidate_edges<'a, W>(
1282 graph: &'a StableDiGraph<W>,
1283 market: &MarketDataView<'_>,
1284 cfg: &CandidateSearchConfig<'_>,
1285 target: NodeIndex,
1286 state: &CandidatePathState<'a, W>,
1287 preferred_only: bool,
1288) -> Vec<ScoredEdge<'a, W>> {
1289 let mut scored = Vec::new();
1290 for edge in graph.edges(state.node) {
1291 let next_node = edge.target();
1292 if !can_extend_path(graph, state, next_node, target, edge.weight(), cfg) {
1293 continue;
1294 }
1295 let priority = match candidate_priority(graph, next_node, target, cfg) {
1296 Some(priority) => priority,
1297 None if preferred_only => continue,
1298 None => 3,
1299 };
1300 let Some(amount_out) = simulate_edge(
1301 market,
1302 &state.amount_out,
1303 &graph[state.node],
1304 edge.weight(),
1305 &graph[next_node],
1306 ) else {
1307 continue;
1308 };
1309 scored.push(ScoredEdge { target: next_node, edge: edge.weight(), amount_out, priority });
1310 }
1311 scored
1312}
1313
1314fn derive_anchor_tokens<W>(graph: &StableDiGraph<W>) -> HashSet<Address> {
1321 let mut by_degree: Vec<(NodeIndex, usize)> = graph
1322 .node_indices()
1323 .map(|node| (node, graph.edges(node).count()))
1324 .collect();
1325 by_degree.sort_unstable_by_key(|(_, degree)| Reverse(*degree));
1326 let mut anchors: HashSet<Address> = by_degree
1327 .into_iter()
1328 .take(DERIVED_ANCHOR_COUNT)
1329 .map(|(node, _)| graph[node].clone())
1330 .collect();
1331 anchors.insert(Address::from([0u8; 20]));
1332 anchors
1333}
1334
1335fn candidate_priority<W>(
1336 graph: &StableDiGraph<W>,
1337 node: NodeIndex,
1338 target: NodeIndex,
1339 cfg: &CandidateSearchConfig<'_>,
1340) -> Option<u8> {
1341 if node == target {
1342 return Some(0);
1343 }
1344 let token = &graph[node];
1345 match cfg.connector_tokens {
1346 Some(tokens) => tokens.contains(token).then_some(1),
1347 None => cfg
1348 .anchor_tokens
1349 .contains(token)
1350 .then_some(2),
1351 }
1352}
1353
1354fn can_extend_path<W>(
1355 graph: &StableDiGraph<W>,
1356 state: &CandidatePathState<'_, W>,
1357 next_node: NodeIndex,
1358 target: NodeIndex,
1359 edge: &EdgeData<W>,
1360 cfg: &CandidateSearchConfig<'_>,
1361) -> bool {
1362 let next_addr = &graph[next_node];
1363 if state
1364 .path
1365 .edge_iter()
1366 .iter()
1367 .any(|existing| existing.component_id == edge.component_id)
1368 {
1369 return false;
1370 }
1371 if state.path.tokens.contains(&next_addr) {
1372 return false;
1373 }
1374 if next_addr == cfg.source_token {
1375 return false;
1376 }
1377 if next_node == target {
1378 return true;
1379 }
1380 cfg.connector_tokens
1381 .map(|tokens| tokens.contains(next_addr))
1382 .unwrap_or(true)
1383}
1384
1385fn simulate_edge<W>(
1386 market: &MarketDataView<'_>,
1387 amount: &BigUint,
1388 token_in_addr: &Address,
1389 edge: &EdgeData<W>,
1390 token_out_addr: &Address,
1391) -> Option<BigUint> {
1392 let token_in = market.get_token(token_in_addr)?;
1393 let token_out = market.get_token(token_out_addr)?;
1394 let state = market.get_simulation_state(&edge.component_id)?;
1395 state
1396 .get_amount_out(amount.clone(), token_in, token_out)
1397 .ok()
1398 .map(|result| result.amount)
1399}
1400
1401fn select_candidate_edges<W>(
1402 mut scored: Vec<ScoredEdge<'_, W>>,
1403 max_edges: usize,
1404) -> Vec<ScoredEdge<'_, W>> {
1405 scored.sort_by(compare_scored_edges);
1406 let mut selected = Vec::new();
1407 let mut per_target: HashMap<NodeIndex, usize> = HashMap::new();
1408 for edge in scored {
1409 let limit = if edge.priority == 0 {
1410 CANDIDATE_DIRECT_EDGES_PER_TOKEN
1411 } else {
1412 CANDIDATE_CONNECTOR_EDGES_PER_TOKEN
1413 };
1414 let count = per_target
1415 .entry(edge.target)
1416 .or_default();
1417 if *count >= limit {
1418 continue;
1419 }
1420 *count += 1;
1421 selected.push(edge);
1422 if selected.len() >= max_edges {
1423 break;
1424 }
1425 }
1426 selected
1427}
1428
1429fn compare_scored_edges<W>(a: &ScoredEdge<'_, W>, b: &ScoredEdge<'_, W>) -> Ordering {
1430 a.priority
1431 .cmp(&b.priority)
1432 .then_with(|| b.amount_out.cmp(&a.amount_out))
1433}
1434
1435fn prune_candidate_frontier<W>(
1436 by_node: HashMap<NodeIndex, Vec<CandidatePathState<'_, W>>>,
1437) -> Vec<CandidatePathState<'_, W>> {
1438 by_node
1439 .into_values()
1440 .flat_map(|mut states| {
1441 states.sort_by(|a, b| b.amount_out.cmp(&a.amount_out));
1442 states.truncate(CANDIDATE_STATES_PER_NODE);
1443 states
1444 })
1445 .collect()
1446}
1447
1448fn rank_found_candidate_paths<'a, W>(
1449 mut found: Vec<(Path<'a, W>, BigUint)>,
1450 max_candidates: usize,
1451 order: &Order,
1452) -> Result<CandidatePathSet<'a, W>, AlgorithmError> {
1453 found.sort_by(|(_, a), (_, b)| b.cmp(a));
1454 let mut keys = HashSet::new();
1455 let mut paths = Vec::new();
1456 let mut scores = Vec::new();
1457
1458 for (path, amount_out) in found {
1459 if !keys.insert(path_key(&path)) {
1460 continue;
1461 }
1462 let idx = paths.len();
1463 paths.push(path);
1464 scores.push((idx, BigInt::from(amount_out)));
1465 if paths.len() >= max_candidates {
1466 break;
1467 }
1468 }
1469
1470 if paths.is_empty() {
1471 return Err(AlgorithmError::NoPath {
1472 from: order.token_in().clone(),
1473 to: order.token_out().clone(),
1474 reason: NoPathReason::NoGraphPath,
1475 });
1476 }
1477 Ok((paths, scores))
1478}
1479
1480#[cfg(test)]
1481mod tests {
1482 use std::time::Duration;
1483
1484 use alloy::primitives::U256;
1485 use num_bigint::BigUint;
1486 use num_traits::ToPrimitive;
1487 use tycho_simulation::evm::protocol::uniswap_v2::state::UniswapV2State;
1488
1489 use super::*;
1490 use crate::{
1491 algorithm::{
1492 split_test_harness::{
1493 evaluate_scenario, optimal_two_pool_output, split_metrics, split_scenarios,
1494 two_equal_weth_usdc, TWO_EQUAL_USDC_RESERVE, TWO_EQUAL_WETH_RESERVE,
1495 },
1496 test_utils::{
1497 addr, setup_market_unweighted, setup_market_weighted_boxed, token_with_decimals,
1498 ConstantProductSim,
1499 },
1500 },
1501 graph::GraphManager,
1502 types::quote::OrderSide,
1503 };
1504
1505 fn config() -> AlgorithmConfig {
1506 config_ms(2000)
1507 }
1508
1509 fn config_ms(ms: u64) -> AlgorithmConfig {
1510 AlgorithmConfig::new(1, 3, Duration::from_millis(ms), None).unwrap()
1511 }
1512
1513 fn whole_weth_order(token_in: &Address, token_out: &Address, weth: u64) -> Order {
1514 Order::new(
1515 token_in.clone(),
1516 token_out.clone(),
1517 BigUint::from(weth) * BigUint::from(10u64).pow(18),
1518 OrderSide::Sell,
1519 addr(0xFF),
1520 )
1521 }
1522
1523 fn v2_pool(reserve_a: u128, reserve_b: u128) -> UniswapV2State {
1524 UniswapV2State::new(
1525 U256::from(reserve_a) * U256::from(10u64).pow(U256::from(18u64)),
1526 U256::from(reserve_b) * U256::from(10u64).pow(U256::from(18u64)),
1527 )
1528 }
1529
1530 fn water_fill_default() -> WaterFillAlgorithm {
1533 WaterFillAlgorithm::with_config(
1534 AlgorithmConfig::new(1, 4, Duration::from_millis(5000), None).unwrap(),
1535 )
1536 .unwrap()
1537 }
1538
1539 #[tokio::test]
1542 async fn test_water_fill_all_scenarios() {
1543 let algo = water_fill_default();
1544 for scenario in split_scenarios::all() {
1545 let name = scenario.name;
1546 let (market, gm) = scenario.build_market_weighted();
1547 let result = evaluate_scenario(&algo, &scenario, market, gm).await;
1548
1549 result.assert_passes_lower_bound();
1550 let tolerance_pct = if name == "DOUBLE_SPLIT" { 10 } else { 5 };
1556 assert!(
1557 result.within_pct_of_optimum(tolerance_pct),
1558 "'{name}': not within {tolerance_pct}% of optimum",
1559 );
1560 let route = result
1561 .route
1562 .as_ref()
1563 .unwrap_or_else(|| panic!("'{name}': expected a route"));
1564 assert!(route.validate().is_ok(), "'{name}': route validation failed");
1565 }
1566 }
1567
1568 #[tokio::test]
1571 async fn test_water_fill_gas_kills_split() {
1572 let scenario = split_scenarios::gas_kills_split();
1573 let (market, gm) = scenario.build_market_weighted();
1574 let result = evaluate_scenario(&water_fill_default(), &scenario, market, gm).await;
1575 assert_eq!(result.path_count, 1, "high gas should prevent splitting");
1576 }
1577
1578 #[tokio::test]
1583 async fn test_split_fills_when_no_single_path_can() {
1584 let a = token_with_decimals(0x01, "A", 18);
1585 let b = token_with_decimals(0x02, "B", 18);
1586 let pool = || {
1589 Box::new(ConstantProductSim {
1590 reserve_0: BigUint::from(800u64) * BigUint::from(10u64).pow(18),
1591 reserve_1: BigUint::from(10_000u64) * BigUint::from(10u64).pow(18),
1592 gas: 50_000,
1593 }) as Box<dyn ProtocolSim>
1594 };
1595 let (market, gm) = setup_market_weighted_boxed(vec![
1596 ("pool_x", &a, &b, pool()),
1597 ("pool_y", &a, &b, pool()),
1598 ]);
1599 let order = Order::new(
1600 a.address.clone(),
1601 b.address.clone(),
1602 BigUint::from(1_000u64) * BigUint::from(10u64).pow(18),
1603 OrderSide::Sell,
1604 addr(0xFF),
1605 );
1606
1607 let single = MostLiquidAlgorithm::with_config(config())
1609 .unwrap()
1610 .find_best_route(gm.graph(), market.clone(), None, None, &order)
1611 .await;
1612 assert!(single.is_err(), "premise: no single path fills the full order");
1613
1614 let split = WaterFillAlgorithm::with_config(config())
1616 .unwrap()
1617 .find_best_route(gm.graph(), market.clone(), None, None, &order)
1618 .await
1619 .expect("water_fill returns a split when no single path fills");
1620 assert!(split.route().swaps().len() >= 2, "expected a split across both pools");
1621 }
1622
1623 #[tokio::test]
1627 async fn test_water_fill_output_near_two_pool_optimum() {
1628 let m = two_equal_weth_usdc(1);
1629 let trade = 500u64;
1630 let order = whole_weth_order(&m.weth, &m.usdc, trade);
1631
1632 let result = WaterFillAlgorithm::with_config(config())
1633 .unwrap()
1634 .find_best_route(
1635 m.weighted.graph(),
1636 m.market.clone(),
1637 None,
1638 Some(m.derived.clone()),
1639 &order,
1640 )
1641 .await
1642 .expect("split solves");
1643 let (_, path_count, gross) = split_metrics(&result, &m.weth, &m.usdc);
1644 assert_eq!(path_count, 2, "the optimum uses both pools");
1645
1646 let reserve_in = TWO_EQUAL_WETH_RESERVE as f64 * 1e18;
1648 let reserve_out = TWO_EQUAL_USDC_RESERVE as f64 * 1e6;
1649 let trade_amount = trade as f64 * 1e18;
1650 let (_, optimum) =
1651 optimal_two_pool_output(reserve_in, reserve_out, reserve_in, reserve_out, trade_amount);
1652
1653 let gross = gross.to_f64().unwrap();
1654 assert!(
1655 gross >= optimum * 0.999 && gross <= optimum * 1.0001,
1656 "split gross {gross} should be within 0.1% of the two-pool optimum {optimum}",
1657 );
1658 }
1659
1660 #[tokio::test]
1667 async fn test_water_fill_no_loss_under_tight_timeout() {
1668 for ms in [1u64, 5, 50] {
1669 let m = two_equal_weth_usdc(1_000_000_000);
1670 let order = whole_weth_order(&m.weth, &m.usdc, 500);
1671
1672 let split = WaterFillAlgorithm::with_config(config_ms(ms))
1673 .unwrap()
1674 .find_best_route(
1675 m.weighted.graph(),
1676 m.market.clone(),
1677 None,
1678 Some(m.derived.clone()),
1679 &order,
1680 )
1681 .await;
1682 let single = MostLiquidAlgorithm::with_config(config_ms(ms))
1683 .unwrap()
1684 .find_best_route(
1685 m.weighted.graph(),
1686 m.market.clone(),
1687 None,
1688 Some(m.derived.clone()),
1689 &order,
1690 )
1691 .await;
1692
1693 let (Ok(split), Ok(single)) = (split, single) else {
1694 continue;
1695 };
1696 let (split_net, _, _) = split_metrics(&split, &m.weth, &m.usdc);
1697 let (single_net, _, _) = split_metrics(&single, &m.weth, &m.usdc);
1698 assert!(
1699 split_net >= single_net,
1700 "split lost to single-path under {ms}ms timeout: split={split_net} single={single_net}",
1701 );
1702 }
1703 }
1704
1705 #[tokio::test]
1711 async fn test_discovery_finds_and_ranks_parallel_pools() {
1712 let link = token_with_decimals(0x01, "LINK", 18);
1713 let weth = token_with_decimals(0x02, "WETH", 18);
1714 let (market, graph_manager) = setup_market_unweighted(vec![
1715 (
1716 "a_weak_link_weth",
1717 &link,
1718 &weth,
1719 Box::new(v2_pool(2_000_000, 264)) as Box<dyn ProtocolSim>,
1720 ),
1721 (
1722 "z_strong_link_weth",
1723 &link,
1724 &weth,
1725 Box::new(v2_pool(2_000_000, 5_700)) as Box<dyn ProtocolSim>,
1726 ),
1727 ]);
1728 let order = Order::new(
1729 link.address.clone(),
1730 weth.address.clone(),
1731 BigUint::from(1_000u64) * BigUint::from(10u64).pow(18),
1732 OrderSide::Sell,
1733 addr(0xFF),
1734 );
1735
1736 let start = Instant::now();
1737 let view = market.read().await;
1738 let (paths, scores) = find_candidate_paths(
1739 graph_manager.graph(),
1740 &view,
1741 &order,
1742 CandidateSearchConfig {
1743 min_hops: 1,
1744 max_hops: 3,
1745 max_candidates: 128,
1746 connector_tokens: None,
1747 anchor_tokens: &HashSet::new(),
1748 source_token: order.token_in(),
1749 start: &start,
1750 timeout_ms: 2000,
1751 },
1752 )
1753 .expect("discovery finds candidates");
1754
1755 assert_eq!(paths.len(), 2, "both parallel pools should be discovered");
1756 let best_path = &paths[scores[0].0];
1758 assert_eq!(
1759 best_path.edge_iter()[0].component_id,
1760 "z_strong_link_weth",
1761 "discovery should rank by simulated output, not topology or edge weights",
1762 );
1763 }
1764
1765 #[tokio::test]
1767 async fn test_discovery_rejects_invalid_hop_configuration() {
1768 let link = token_with_decimals(0x01, "LINK", 18);
1769 let weth = token_with_decimals(0x02, "WETH", 18);
1770 let (market, graph_manager) = setup_market_unweighted(vec![(
1771 "link_weth",
1772 &link,
1773 &weth,
1774 Box::new(v2_pool(2_000_000, 5_700)) as Box<dyn ProtocolSim>,
1775 )]);
1776 let order = Order::new(
1777 link.address.clone(),
1778 weth.address.clone(),
1779 BigUint::from(1_000u64) * BigUint::from(10u64).pow(18),
1780 OrderSide::Sell,
1781 addr(0xFF),
1782 );
1783
1784 let start = Instant::now();
1785 let view = market.read().await;
1786 let result = find_candidate_paths(
1787 graph_manager.graph(),
1788 &view,
1789 &order,
1790 CandidateSearchConfig {
1791 min_hops: 0,
1792 max_hops: 3,
1793 max_candidates: 128,
1794 connector_tokens: None,
1795 anchor_tokens: &HashSet::new(),
1796 source_token: order.token_in(),
1797 start: &start,
1798 timeout_ms: 2000,
1799 },
1800 );
1801 assert!(
1802 matches!(result, Err(AlgorithmError::InvalidConfiguration { .. })),
1803 "min_hops of 0 should be rejected",
1804 );
1805 }
1806
1807 #[test]
1810 fn test_derive_anchor_tokens_ranks_hub_and_includes_native_sentinel() {
1811 let hub = token_with_decimals(0x01, "HUB", 18);
1812 let a = token_with_decimals(0x02, "A", 18);
1813 let b = token_with_decimals(0x03, "B", 18);
1814 let c = token_with_decimals(0x04, "C", 18);
1815 let (_market, graph_manager) = setup_market_unweighted(vec![
1816 ("hub_a", &hub, &a, Box::new(v2_pool(1, 1)) as Box<dyn ProtocolSim>),
1817 ("hub_b", &hub, &b, Box::new(v2_pool(1, 1)) as Box<dyn ProtocolSim>),
1818 ("hub_c", &hub, &c, Box::new(v2_pool(1, 1)) as Box<dyn ProtocolSim>),
1819 ]);
1820
1821 let anchors = derive_anchor_tokens(graph_manager.graph());
1822
1823 assert!(anchors.contains(&hub.address), "highest-degree token should be anchored");
1824 assert!(
1825 anchors.contains(&Address::from([0u8; 20])),
1826 "native-ETH sentinel should always be anchored",
1827 );
1828 }
1829}