use std::{
cmp::{Ordering, Reverse},
collections::{HashMap, HashSet},
time::{Duration, Instant},
};
use num_bigint::{BigInt, BigUint};
use num_traits::Zero;
use petgraph::{graph::NodeIndex, prelude::EdgeRef};
use tracing::{debug, instrument};
use tycho_simulation::{
tycho_common::simulation::protocol_sim::ProtocolSim, tycho_core::models::Address,
};
use super::{
most_liquid::DepthAndPrice,
split_primitives::{build_split_route, HopDescriptor, PathAllocation, SimulatedHop},
Algorithm, AlgorithmConfig, MostLiquidAlgorithm, NoPathReason,
};
use crate::{
derived::{computation::ComputationRequirements, types::TokenGasPrices, SharedDerivedDataRef},
feed::market_data::{MarketData, MarketDataView, MarketState, StateLabel},
graph::{petgraph::StableDiGraph, EdgeData, Path, PetgraphStableDiGraphManager},
types::{ComponentId, Order, Route, RouteResult},
AlgorithmError,
};
const DEFAULT_MAX_CANDIDATES: usize = 5000;
const BOUNDED_DISCOVERY_CANDIDATES: usize = 128;
const DEFAULT_MAX_PATHS: usize = 4;
const COARSE_CHUNKS: usize = 20;
const FINE_CHUNKS: usize = 256;
const SHARED_FULL_PATHS: usize = 8;
const SHARED_MARGIN_PROBE_PATHS: usize = 32;
const SHARED_MARGIN_PATHS: usize = 8;
const SHARED_MAX_CANDIDATES: usize = 12;
const CANDIDATE_STATES_PER_NODE: usize = 4;
const CANDIDATE_EDGES_PER_STATE: usize = 16;
const CANDIDATE_DIRECT_EDGES_PER_TOKEN: usize = 4;
const CANDIDATE_CONNECTOR_EDGES_PER_TOKEN: usize = 2;
const DERIVED_ANCHOR_COUNT: usize = 16;
const EXCHANGE_DELTA_FLOOR: usize = 64;
const EXCHANGE_MAX_SIMS: usize = 400;
struct SplitCandidate {
route: Route,
gross: BigUint,
gas: BigUint,
}
impl SplitCandidate {
fn net(
&self,
gas_price: &BigUint,
token_prices: Option<&TokenGasPrices>,
token_out: &Address,
) -> BigInt {
let cost =
WaterFillAlgorithm::gas_cost_in_token(&self.gas, gas_price, token_prices, token_out);
match cost {
Some(c) => BigInt::from(self.gross.clone()) - BigInt::from(c),
None => BigInt::from(self.gross.clone()),
}
}
}
pub struct WaterFillAlgorithm {
min_hops: usize,
max_hops: usize,
timeout: Duration,
max_candidates: usize,
max_paths: usize,
connector_tokens: Option<HashSet<Address>>,
}
struct ExchangeMove {
donor: usize,
recipient: usize,
donor_net: BigInt,
recip_net: BigInt,
gain: BigInt,
}
struct StepResult {
amount_out: BigUint,
gas: BigUint,
new_states: Vec<(ComponentId, Box<dyn ProtocolSim>)>,
}
struct SplitContext<'a, 'g> {
ordered: &'a [Path<'g, DepthAndPrice>],
market: &'a MarketState,
gas_price: &'a BigUint,
token_prices: Option<&'a TokenGasPrices>,
order: &'a Order,
start: Instant,
}
struct SetupResult<'a> {
ordered: Vec<Path<'a, DepthAndPrice>>,
market: MarketState,
gas_price: BigUint,
best_single: Option<RouteResult>,
token_prices: Option<TokenGasPrices>,
}
impl WaterFillAlgorithm {
pub(crate) fn with_config(config: AlgorithmConfig) -> Result<Self, AlgorithmError> {
Ok(Self {
min_hops: config.min_hops(),
max_hops: config.max_hops(),
timeout: config.timeout(),
max_candidates: config
.max_routes()
.unwrap_or(DEFAULT_MAX_CANDIDATES)
.max(DEFAULT_MAX_PATHS),
max_paths: DEFAULT_MAX_PATHS,
connector_tokens: config.connector_tokens().cloned(),
})
}
fn simulate_step(
path: &Path<DepthAndPrice>,
market: &MarketState,
overlay: &HashMap<ComponentId, Box<dyn ProtocolSim>>,
amount: BigUint,
) -> Option<StepResult> {
let mut current = amount;
let mut total_gas = BigUint::zero();
let path_reuses_pool = {
let mut seen: HashSet<&ComponentId> = HashSet::with_capacity(path.len());
!path
.edge_iter()
.iter()
.all(|e| seen.insert(&e.component_id))
};
let mut intra_path_states: HashMap<ComponentId, Box<dyn ProtocolSim>> = HashMap::new();
let mut new_states: Vec<(ComponentId, Box<dyn ProtocolSim>)> =
Vec::with_capacity(path.len());
for (address_in, edge, address_out) in path.iter() {
let token_in = market.get_token(address_in)?;
let token_out = market.get_token(address_out)?;
let component_id = &edge.component_id;
let state = intra_path_states
.get(component_id)
.map(Box::as_ref)
.or_else(|| {
overlay
.get(component_id)
.map(Box::as_ref)
})
.or_else(|| market.get_simulation_state(component_id))?;
let result = state
.get_amount_out(current.clone(), token_in, token_out)
.ok()?;
total_gas += &result.gas;
if path_reuses_pool {
intra_path_states.insert(component_id.clone(), result.new_state.clone_box());
}
new_states.push((component_id.clone(), result.new_state));
current = result.amount;
}
Some(StepResult { amount_out: current, gas: total_gas, new_states })
}
fn gas_cost_in_token(
total_gas: &BigUint,
gas_price_wei: &BigUint,
token_prices: Option<&TokenGasPrices>,
token_out: &Address,
) -> Option<BigUint> {
let price = token_prices?.get(token_out)?;
if price.denominator.is_zero() {
return None;
}
Some(total_gas * gas_price_wei * &price.numerator / &price.denominator)
}
fn activation_cost(ctx: &SplitContext, gas: &BigUint) -> BigInt {
Self::gas_cost_in_token(gas, ctx.gas_price, ctx.token_prices, ctx.order.token_out())
.map(BigInt::from)
.unwrap_or_else(BigInt::zero)
}
fn select_disjoint(ranked: &[Path<DepthAndPrice>], max_paths: usize) -> Vec<usize> {
let mut visited_pools: HashSet<&ComponentId> = HashSet::new();
let mut selected = Vec::new();
for (idx, path) in ranked.iter().enumerate() {
let path_pools: Vec<&ComponentId> = path
.edge_iter()
.iter()
.map(|e| &e.component_id)
.collect();
if path_pools
.iter()
.any(|c| visited_pools.contains(*c))
{
continue;
}
for c in path_pools {
visited_pools.insert(c);
}
selected.push(idx);
if selected.len() >= max_paths {
break;
}
}
selected
}
#[instrument(level = "debug", skip_all)]
async fn setup<'a>(
&self,
graph: &'a StableDiGraph<DepthAndPrice>,
market: MarketData,
label: Option<StateLabel>,
derived: Option<SharedDerivedDataRef>,
order: &Order,
start: Instant,
) -> Result<SetupResult<'a>, AlgorithmError> {
let token_prices = if let Some(ref derived) = derived {
derived
.read()
.await
.token_prices()
.cloned()
} else {
None
};
let all_paths = MostLiquidAlgorithm::find_paths(
graph,
order.token_in(),
order.token_out(),
self.min_hops,
self.max_hops,
self.connector_tokens.as_ref(),
)?;
if all_paths.is_empty() {
return Err(AlgorithmError::NoPath {
from: order.token_in().clone(),
to: order.token_out().clone(),
reason: NoPathReason::NoGraphPath,
});
}
let mut scored: Vec<(Path<DepthAndPrice>, f64)> = all_paths
.into_iter()
.map(|p| {
let s = MostLiquidAlgorithm::try_score_path(&p).unwrap_or(f64::MIN);
(p, s)
})
.collect();
scored.sort_by(|(_, a), (_, b)| {
b.partial_cmp(a)
.unwrap_or(std::cmp::Ordering::Equal)
});
scored.truncate(self.max_candidates);
let mut paths: Vec<Path<DepthAndPrice>> = scored
.into_iter()
.map(|(p, _)| p)
.collect();
let timeout_ms = self.timeout.as_millis() as u64;
let market = {
let view = match label.as_ref() {
Some(l) => market
.read_labeled(l)
.await
.map_err(|e| AlgorithmError::Other(e.to_string()))?,
None => market.read().await,
};
if view.gas_price().is_none() {
return Err(AlgorithmError::DataNotFound { kind: "gas price", id: None });
}
let anchor_tokens = derive_anchor_tokens(graph);
let bounded = find_candidate_paths(
graph,
&view,
order,
CandidateSearchConfig {
min_hops: self.min_hops,
max_hops: self.max_hops,
max_candidates: BOUNDED_DISCOVERY_CANDIDATES,
connector_tokens: self.connector_tokens.as_ref(),
anchor_tokens: &anchor_tokens,
source_token: order.token_in(),
start: &start,
timeout_ms,
},
);
match bounded {
Ok((bounded_paths, _)) => {
let mut keys: HashSet<Vec<ComponentId>> = paths.iter().map(path_key).collect();
let mut union = Vec::with_capacity(bounded_paths.len() + paths.len());
for path in bounded_paths {
if keys.insert(path_key(&path)) {
union.push(path);
}
}
union.append(&mut paths);
paths = union;
}
Err(e) => {
debug!(error = %e, "water-fill bounded discovery failed; using exhaustive candidates only")
}
}
let component_ids: HashSet<ComponentId> = paths
.iter()
.flat_map(|p| {
p.edge_iter()
.iter()
.map(|e| e.component_id.clone())
})
.collect();
let subset = view.extract_subset_with_overlay(&component_ids);
drop(view);
subset
};
let gas_price = market
.gas_price()
.ok_or(AlgorithmError::DataNotFound { kind: "gas price", id: None })?
.effective_gas_price()
.clone();
let amount_in = order.amount().clone();
let mut best_single: Option<RouteResult> = None;
let mut full_outputs: Vec<(usize, BigUint)> = Vec::new();
for (idx, path) in paths.iter().enumerate() {
if start.elapsed().as_millis() as u64 > timeout_ms {
break;
}
match MostLiquidAlgorithm::simulate_path(
path,
&market,
token_prices.as_ref(),
amount_in.clone(),
) {
Ok(result) => {
let gross = result
.route()
.swaps()
.last()
.map(|s| s.amount_out().clone())
.unwrap_or_else(BigUint::zero);
full_outputs.push((idx, gross));
if best_single
.as_ref()
.map(|b| result.net_amount_out() > b.net_amount_out())
.unwrap_or(true)
{
best_single = Some(result);
}
}
Err(_) => full_outputs.push((idx, BigUint::zero())),
}
}
full_outputs.sort_by(|(_, a), (_, b)| b.cmp(a));
let ordered: Vec<Path<DepthAndPrice>> = full_outputs
.into_iter()
.map(|(idx, _)| paths[idx].clone())
.collect();
debug!(
candidate_paths = ordered.len(),
elapsed_ms = start.elapsed().as_millis(),
"water-fill discovery + full-amount ranking"
);
Ok(SetupResult { ordered, market, gas_price, best_single, token_prices })
}
}
impl Algorithm for WaterFillAlgorithm {
type GraphType = StableDiGraph<DepthAndPrice>;
type GraphManager = PetgraphStableDiGraphManager<DepthAndPrice>;
fn name(&self) -> &str {
"water_fill"
}
#[instrument(level = "debug", skip_all, fields(order_id = %order.id()))]
async fn find_best_route(
&self,
graph: &Self::GraphType,
market: MarketData,
label: Option<StateLabel>,
derived: Option<SharedDerivedDataRef>,
order: &Order,
) -> Result<RouteResult, AlgorithmError> {
let start = Instant::now();
if !order.is_sell() {
return Err(AlgorithmError::ExactOutNotSupported);
}
let SetupResult { ordered, market, gas_price, best_single, token_prices } = self
.setup(graph, market, label, derived, order, start)
.await?;
let token_out = order.token_out();
let ctx = SplitContext {
ordered: &ordered,
market: &market,
gas_price: &gas_price,
token_prices: token_prices.as_ref(),
order,
start,
};
let mut candidates: Vec<SplitCandidate> = Vec::new();
let disjoint = Self::select_disjoint(ctx.ordered, self.max_paths);
let coarse = (disjoint.len() >= 2)
.then(|| self.disjoint_waterfill(&ctx, &disjoint, COARSE_CHUNKS, true))
.flatten();
if let Some(coarse) = coarse.as_deref() {
if let Some(c) = self.build_disjoint_legs(&ctx, &disjoint, coarse) {
candidates.push(c);
}
if let Some(c) = self.disjoint_refine(&ctx, &disjoint, coarse, FINE_CHUNKS) {
candidates.push(c);
}
}
if let Some(c) = self.fillspill_alloc(&ctx, FINE_CHUNKS) {
candidates.push(c);
}
let candidate_count = candidates.len();
let baseline_net = best_single
.as_ref()
.map(|b| b.net_amount_out().clone());
let mut best: Option<(BigInt, SplitCandidate)> = None;
for cand in candidates {
let net = cand.net(&gas_price, token_prices.as_ref(), token_out);
let beats = match (&best, &baseline_net) {
(Some((current, _)), _) => net > *current,
(None, Some(base)) => net > *base,
(None, None) => true,
};
if beats {
best = Some((net, cand));
}
}
let split_won = best.is_some();
debug!(
candidate_count,
split_won,
elapsed_ms = start.elapsed().as_millis(),
"water-fill selected {}",
if split_won { "split candidate" } else { "single path" }
);
match best {
Some((net, cand)) => Ok(RouteResult::new(cand.route, net, gas_price)),
None => best_single.ok_or(AlgorithmError::InsufficientLiquidity),
}
}
fn computation_requirements(&self) -> ComputationRequirements {
ComputationRequirements::none()
.allow_stale("token_prices")
.expect("Conflicting Computation Requirements")
}
fn timeout(&self) -> Duration {
self.timeout
}
}
impl WaterFillAlgorithm {
fn disjoint_refine(
&self,
ctx: &SplitContext,
disjoint: &[usize],
coarse: &[BigUint],
fine_chunks: usize,
) -> Option<SplitCandidate> {
let active: Vec<usize> = disjoint
.iter()
.copied()
.zip(coarse.iter())
.filter(|(_, amt)| !amt.is_zero())
.map(|(idx, _)| idx)
.collect();
if active.is_empty() {
return None;
}
let fine = self.disjoint_waterfill(ctx, &active, fine_chunks, false)?;
let refined = self.disjoint_exchange(ctx, &active, fine_chunks, fine);
self.build_disjoint_legs(ctx, &active, &refined)
}
fn path_net(
ctx: &SplitContext,
path: &Path<DepthAndPrice>,
amount: &BigUint,
) -> Option<BigInt> {
if amount.is_zero() {
return Some(BigInt::zero());
}
let empty: HashMap<ComponentId, Box<dyn ProtocolSim>> = HashMap::new();
let step = Self::simulate_step(path, ctx.market, &empty, amount.clone())?;
let activation = Self::activation_cost(ctx, &step.gas);
Some(BigInt::from(step.amount_out) - activation)
}
fn disjoint_exchange(
&self,
ctx: &SplitContext,
active: &[usize],
fine_chunks: usize,
alloc: Vec<BigUint>,
) -> Vec<BigUint> {
let path_count = active.len();
if path_count < 2 {
return alloc;
}
let amount_in = ctx.order.amount().clone();
let fine_chunks = fine_chunks.max(1);
let mut delta = &amount_in / fine_chunks;
let min_delta = &amount_in / (fine_chunks * EXCHANGE_DELTA_FLOOR);
if delta.is_zero() {
return alloc;
}
let timeout_ms = self.timeout.as_millis() as u64;
let mut cum = alloc;
let mut net_cache: Vec<BigInt> = Vec::with_capacity(path_count);
for (i, &path_idx) in active.iter().enumerate() {
let Some(net) = Self::path_net(ctx, &ctx.ordered[path_idx], &cum[i]) else {
return cum;
};
net_cache.push(net);
}
let mut sims = 0usize;
while delta >= min_delta && !delta.is_zero() {
if ctx.start.elapsed().as_millis() as u64 > timeout_ms || sims >= EXCHANGE_MAX_SIMS {
break;
}
let mut best: Option<ExchangeMove> = None;
for donor in 0..path_count {
if sims >= EXCHANGE_MAX_SIMS {
break;
}
if cum[donor] < delta {
continue;
}
let donor_amt = &cum[donor] - δ
let Some(donor_net) = Self::path_net(ctx, &ctx.ordered[active[donor]], &donor_amt)
else {
continue;
};
sims += 1;
for recipient in 0..path_count {
if recipient == donor || sims >= EXCHANGE_MAX_SIMS {
continue;
}
let recip_amt = &cum[recipient] + δ
let Some(recip_net) =
Self::path_net(ctx, &ctx.ordered[active[recipient]], &recip_amt)
else {
continue;
};
sims += 1;
let before = &net_cache[donor] + &net_cache[recipient];
let after = &donor_net + &recip_net;
if after <= before {
continue;
}
let gain = after - before;
if best
.as_ref()
.map(|m| gain > m.gain)
.unwrap_or(true)
{
best = Some(ExchangeMove {
donor,
recipient,
donor_net: donor_net.clone(),
recip_net,
gain,
});
}
}
}
let Some(mv) = best else {
delta = &delta / 2usize;
continue;
};
cum[mv.donor] = &cum[mv.donor] - δ
cum[mv.recipient] = &cum[mv.recipient] + δ
net_cache[mv.donor] = mv.donor_net;
net_cache[mv.recipient] = mv.recip_net;
}
cum
}
fn allocation_commit(
path: &Path<DepthAndPrice>,
market: &MarketState,
overrides: &mut HashMap<ComponentId, Box<dyn ProtocolSim>>,
amount: BigUint,
flow_fraction: f64,
) -> Option<PathAllocation> {
let amount_in = amount;
let mut current = amount_in.clone();
let mut hops = Vec::with_capacity(path.len());
for (address_in, edge, address_out) in path.iter() {
let token_in = market.get_token(address_in)?;
let token_out = market.get_token(address_out)?;
let component_id = &edge.component_id;
let state = overrides
.get(component_id)
.map(Box::as_ref)
.or_else(|| market.get_simulation_state(component_id))?;
let result = state
.get_amount_out(current.clone(), token_in, token_out)
.ok()?;
hops.push(SimulatedHop {
descriptor: HopDescriptor::new(
component_id.clone(),
token_in.clone(),
token_out.clone(),
),
amount_out: result.amount.clone(),
gas: result.gas.clone(),
});
overrides.insert(component_id.clone(), result.new_state);
current = result.amount;
}
Some(PathAllocation {
hops,
flow_fraction,
amount_in,
amount_out: current,
marginal_price_product: 0.0,
})
}
fn candidate_from_allocations(
ctx: &SplitContext,
allocations: &[PathAllocation],
) -> Option<SplitCandidate> {
let route = build_split_route(allocations, ctx.market, ctx.order).ok()?;
let token_out = ctx.order.token_out();
let gross = route
.swaps()
.iter()
.filter(|s| s.token_out() == token_out)
.fold(BigUint::zero(), |acc, s| acc + s.amount_out());
if gross.is_zero() {
return None;
}
let gas = route.total_gas();
Some(SplitCandidate { route, gross, gas })
}
fn build_disjoint_legs(
&self,
ctx: &SplitContext,
subset: &[usize],
alloc: &[BigUint],
) -> Option<SplitCandidate> {
let amount_in = ctx.order.amount().clone();
let mut allocations = Vec::new();
for (i, &path_idx) in subset.iter().enumerate() {
if alloc[i].is_zero() {
continue;
}
let mut overrides: HashMap<ComponentId, Box<dyn ProtocolSim>> = HashMap::new();
let allocation = Self::allocation_commit(
&ctx.ordered[path_idx],
ctx.market,
&mut overrides,
alloc[i].clone(),
ratio(&alloc[i], &amount_in),
)?;
allocations.push(allocation);
}
if allocations.is_empty() {
return None;
}
Self::candidate_from_allocations(ctx, &allocations)
}
fn disjoint_waterfill(
&self,
ctx: &SplitContext,
subset: &[usize],
num_chunks: usize,
gate: bool,
) -> Option<Vec<BigUint>> {
let amount_in = ctx.order.amount().clone();
let num_chunks = num_chunks.max(1);
let base_chunk = &amount_in / num_chunks;
if base_chunk.is_zero() {
return None;
}
let remainder = &amount_in - &base_chunk * num_chunks;
let timeout_ms = self.timeout.as_millis() as u64;
let path_count = subset.len();
let mut committed: Vec<HashMap<ComponentId, Box<dyn ProtocolSim>>> = (0..path_count)
.map(|_| HashMap::new())
.collect();
let mut cum_in: Vec<BigUint> = vec![BigUint::zero(); path_count];
let mut activated: Vec<bool> = vec![!gate; path_count];
for chunk_idx in 0..num_chunks {
if ctx.start.elapsed().as_millis() as u64 > timeout_ms {
break;
}
let chunk = if chunk_idx == 0 { &base_chunk + &remainder } else { base_chunk.clone() };
let mut best: Option<(usize, BigInt, StepResult)> = None;
for (i, &path_idx) in subset.iter().enumerate() {
let Some(step) = Self::simulate_step(
&ctx.ordered[path_idx],
ctx.market,
&committed[i],
chunk.clone(),
) else {
continue;
};
let gross_marginal = BigInt::from(step.amount_out.clone());
let net_marginal = if activated[i] {
gross_marginal
} else {
let activation = Self::activation_cost(ctx, &step.gas);
gross_marginal - activation
};
if best
.as_ref()
.map(|(_, m, _)| &net_marginal > m)
.unwrap_or(true)
{
best = Some((i, net_marginal, step));
}
}
let Some((best_i, _, step)) = best else {
break;
};
for (id, state) in step.new_states {
committed[best_i].insert(id, state);
}
cum_in[best_i] += &chunk;
activated[best_i] = true;
}
Some(cum_in)
}
fn select_shared_candidates(&self, ctx: &SplitContext) -> Vec<usize> {
let mut candidates: Vec<usize> = (0..ctx.ordered.len().min(SHARED_FULL_PATHS)).collect();
let first_chunk = ctx.order.amount() / COARSE_CHUNKS;
if first_chunk.is_zero() {
return candidates;
}
let timeout_ms = self.timeout.as_millis() as u64;
let empty: HashMap<ComponentId, Box<dyn ProtocolSim>> = HashMap::new();
let mut marginal: Vec<(usize, BigInt)> = Vec::new();
for (idx, path) in ctx
.ordered
.iter()
.enumerate()
.take(SHARED_MARGIN_PROBE_PATHS)
{
if ctx.start.elapsed().as_millis() as u64 > timeout_ms {
break;
}
let Some(step) = Self::simulate_step(path, ctx.market, &empty, first_chunk.clone())
else {
continue;
};
let activation = Self::activation_cost(ctx, &step.gas);
marginal.push((idx, BigInt::from(step.amount_out) - activation));
}
marginal.sort_by(|(_, a), (_, b)| b.cmp(a));
for (idx, net) in marginal
.into_iter()
.take(SHARED_MARGIN_PATHS)
{
if net <= BigInt::zero() {
continue;
}
if !candidates.contains(&idx) {
candidates.push(idx);
}
if candidates.len() >= SHARED_MAX_CANDIDATES {
break;
}
}
candidates
}
fn fillspill_alloc(&self, ctx: &SplitContext, fine_chunks: usize) -> Option<SplitCandidate> {
let candidates = self.select_shared_candidates(ctx);
if candidates.len() < 2 {
return None;
}
let (coarse_counts, _) = self.fillspill_waterfill(ctx, &candidates, COARSE_CHUNKS, true)?;
let active: Vec<usize> = candidates
.iter()
.copied()
.zip(coarse_counts.iter())
.filter(|(_, count)| **count > 0)
.map(|(idx, _)| idx)
.collect();
if active.len() < 2 {
return None;
}
let (_, schedule) = self.fillspill_waterfill(ctx, &active, fine_chunks, false)?;
if schedule.is_empty() {
return None;
}
self.build_fillspill_route(ctx, &active, &schedule)
}
#[allow(clippy::type_complexity)]
fn fillspill_waterfill(
&self,
ctx: &SplitContext,
subset: &[usize],
num_chunks: usize,
gate: bool,
) -> Option<(Vec<usize>, Vec<(usize, BigUint)>)> {
let amount_in = ctx.order.amount().clone();
let num_chunks = num_chunks.max(1);
let base_chunk = &amount_in / num_chunks;
if base_chunk.is_zero() {
return None;
}
let remainder = &amount_in - &base_chunk * num_chunks;
let timeout_ms = self.timeout.as_millis() as u64;
let mut overlay: HashMap<ComponentId, Box<dyn ProtocolSim>> = HashMap::new();
let mut activated: Vec<bool> = vec![!gate; subset.len()];
let mut active_count = if gate { 0 } else { subset.len() };
let mut counts: Vec<usize> = vec![0; subset.len()];
let mut schedule: Vec<(usize, BigUint)> = Vec::with_capacity(num_chunks);
for chunk_idx in 0..num_chunks {
if ctx.start.elapsed().as_millis() as u64 > timeout_ms {
break;
}
let chunk = if chunk_idx == 0 { &base_chunk + &remainder } else { base_chunk.clone() };
let mut best: Option<(usize, BigInt, StepResult)> = None;
for (i, &path_idx) in subset.iter().enumerate() {
if !activated[i] && active_count >= self.max_paths {
continue;
}
let Some(step) = Self::simulate_step(
&ctx.ordered[path_idx],
ctx.market,
&overlay,
chunk.clone(),
) else {
continue;
};
let gross_marginal = BigInt::from(step.amount_out.clone());
let net_marginal = if activated[i] {
gross_marginal
} else {
let activation = Self::activation_cost(ctx, &step.gas);
gross_marginal - activation
};
if best
.as_ref()
.map(|(_, m, _)| &net_marginal > m)
.unwrap_or(true)
{
best = Some((i, net_marginal, step));
}
}
let Some((best_i, _, step)) = best else {
break;
};
for (id, state) in step.new_states {
overlay.insert(id, state);
}
if !activated[best_i] {
activated[best_i] = true;
active_count += 1;
}
counts[best_i] += 1;
schedule.push((best_i, chunk));
}
Some((counts, schedule))
}
fn build_fillspill_route(
&self,
ctx: &SplitContext,
active: &[usize],
schedule: &[(usize, BigUint)],
) -> Option<SplitCandidate> {
let amount_in = ctx.order.amount().clone();
let mut cand_in: Vec<BigUint> = vec![BigUint::zero(); active.len()];
for (i, chunk) in schedule {
cand_in[*i] += chunk;
}
let mut execution_order: Vec<usize> = (0..active.len())
.filter(|&i| !cand_in[i].is_zero())
.collect();
if execution_order.len() < 2 {
return None;
}
execution_order.sort_by(|&a, &b| cand_in[b].cmp(&cand_in[a]));
let mut overrides: HashMap<ComponentId, Box<dyn ProtocolSim>> = HashMap::new();
let mut allocations = Vec::new();
for i in execution_order {
let allocation = Self::allocation_commit(
&ctx.ordered[active[i]],
ctx.market,
&mut overrides,
cand_in[i].clone(),
ratio(&cand_in[i], &amount_in),
)?;
allocations.push(allocation);
}
Self::candidate_from_allocations(ctx, &allocations)
}
}
fn path_key<W>(path: &Path<'_, W>) -> Vec<ComponentId> {
path.edge_iter()
.iter()
.map(|e| e.component_id.clone())
.collect()
}
fn ratio(numerator: &BigUint, denominator: &BigUint) -> f64 {
use num_traits::ToPrimitive;
let n = numerator.to_f64().unwrap_or(0.0);
let d = denominator.to_f64().unwrap_or(1.0);
if d == 0.0 {
0.0
} else {
n / d
}
}
type RankedPathScores = Vec<(usize, BigInt)>;
type CandidatePathSet<'a, W = ()> = (Vec<Path<'a, W>>, RankedPathScores);
#[derive(Clone)]
struct CandidatePathState<'a, W> {
node: NodeIndex,
path: Path<'a, W>,
amount_out: BigUint,
}
struct ScoredEdge<'a, W> {
target: NodeIndex,
edge: &'a EdgeData<W>,
amount_out: BigUint,
priority: u8,
}
#[derive(Clone, Copy)]
struct CandidateSearchConfig<'a> {
min_hops: usize,
max_hops: usize,
max_candidates: usize,
connector_tokens: Option<&'a HashSet<Address>>,
anchor_tokens: &'a HashSet<Address>,
source_token: &'a Address,
start: &'a Instant,
timeout_ms: u64,
}
fn timed_out(start: &Instant, timeout_ms: u64) -> bool {
start.elapsed().as_millis() as u64 > timeout_ms
}
fn find_candidate_paths<'a, W>(
graph: &'a StableDiGraph<W>,
market: &MarketDataView<'_>,
order: &Order,
cfg: CandidateSearchConfig<'_>,
) -> Result<CandidatePathSet<'a, W>, AlgorithmError>
where
W: Clone,
{
if cfg.min_hops == 0 || cfg.min_hops > cfg.max_hops {
return Err(AlgorithmError::InvalidConfiguration {
reason: format!(
"invalid hop configuration: min_hops={} max_hops={}",
cfg.min_hops, cfg.max_hops,
),
});
}
let from_idx =
find_token_node(graph, order.token_in(), NoPathReason::SourceTokenNotInGraph, order)?;
let to_idx =
find_token_node(graph, order.token_out(), NoPathReason::DestinationTokenNotInGraph, order)?;
let mut found = Vec::new();
let mut frontier = vec![CandidatePathState {
node: from_idx,
path: Path::new(),
amount_out: order.amount().clone(),
}];
for _depth in 0..cfg.max_hops {
if timed_out(cfg.start, cfg.timeout_ms) || frontier.is_empty() {
break;
}
let mut next_by_node: HashMap<NodeIndex, Vec<CandidatePathState<'a, W>>> = HashMap::new();
for state in frontier {
if state.node == to_idx && from_idx != to_idx {
continue;
}
expand_candidate_state(
graph,
market,
&cfg,
to_idx,
state,
&mut found,
&mut next_by_node,
);
}
frontier = prune_candidate_frontier(next_by_node);
}
rank_found_candidate_paths(found, cfg.max_candidates, order)
}
fn find_token_node<W>(
graph: &StableDiGraph<W>,
token: &Address,
reason: NoPathReason,
order: &Order,
) -> Result<NodeIndex, AlgorithmError> {
graph
.node_indices()
.find(|&node| &graph[node] == token)
.ok_or(AlgorithmError::NoPath {
from: order.token_in().clone(),
to: order.token_out().clone(),
reason,
})
}
fn expand_candidate_state<'a, W>(
graph: &'a StableDiGraph<W>,
market: &MarketDataView<'_>,
cfg: &CandidateSearchConfig<'_>,
target: NodeIndex,
state: CandidatePathState<'a, W>,
found: &mut Vec<(Path<'a, W>, BigUint)>,
next_by_node: &mut HashMap<NodeIndex, Vec<CandidatePathState<'a, W>>>,
) where
W: Clone,
{
let edges = candidate_edges_for_state(graph, market, cfg, target, &state);
for candidate in edges {
if timed_out(cfg.start, cfg.timeout_ms) {
break;
}
let mut path = state.path.clone();
path.add_hop(&graph[state.node], candidate.edge, &graph[candidate.target]);
let path_state = CandidatePathState {
node: candidate.target,
path: path.clone(),
amount_out: candidate.amount_out,
};
if candidate.target == target && path.len() >= cfg.min_hops {
found.push((path.clone(), path_state.amount_out.clone()));
}
if path.len() < cfg.max_hops {
next_by_node
.entry(candidate.target)
.or_default()
.push(path_state);
}
}
}
fn candidate_edges_for_state<'a, W>(
graph: &'a StableDiGraph<W>,
market: &MarketDataView<'_>,
cfg: &CandidateSearchConfig<'_>,
target: NodeIndex,
state: &CandidatePathState<'a, W>,
) -> Vec<ScoredEdge<'a, W>> {
let mut preferred = score_candidate_edges(graph, market, cfg, target, state, true);
if preferred.is_empty() {
preferred = score_candidate_edges(graph, market, cfg, target, state, false);
}
select_candidate_edges(preferred, CANDIDATE_EDGES_PER_STATE)
}
fn score_candidate_edges<'a, W>(
graph: &'a StableDiGraph<W>,
market: &MarketDataView<'_>,
cfg: &CandidateSearchConfig<'_>,
target: NodeIndex,
state: &CandidatePathState<'a, W>,
preferred_only: bool,
) -> Vec<ScoredEdge<'a, W>> {
let mut scored = Vec::new();
for edge in graph.edges(state.node) {
let next_node = edge.target();
if !can_extend_path(graph, state, next_node, target, edge.weight(), cfg) {
continue;
}
let priority = match candidate_priority(graph, next_node, target, cfg) {
Some(priority) => priority,
None if preferred_only => continue,
None => 3,
};
let Some(amount_out) = simulate_edge(
market,
&state.amount_out,
&graph[state.node],
edge.weight(),
&graph[next_node],
) else {
continue;
};
scored.push(ScoredEdge { target: next_node, edge: edge.weight(), amount_out, priority });
}
scored
}
fn derive_anchor_tokens<W>(graph: &StableDiGraph<W>) -> HashSet<Address> {
let mut by_degree: Vec<(NodeIndex, usize)> = graph
.node_indices()
.map(|node| (node, graph.edges(node).count()))
.collect();
by_degree.sort_unstable_by_key(|(_, degree)| Reverse(*degree));
let mut anchors: HashSet<Address> = by_degree
.into_iter()
.take(DERIVED_ANCHOR_COUNT)
.map(|(node, _)| graph[node].clone())
.collect();
anchors.insert(Address::from([0u8; 20]));
anchors
}
fn candidate_priority<W>(
graph: &StableDiGraph<W>,
node: NodeIndex,
target: NodeIndex,
cfg: &CandidateSearchConfig<'_>,
) -> Option<u8> {
if node == target {
return Some(0);
}
let token = &graph[node];
match cfg.connector_tokens {
Some(tokens) => tokens.contains(token).then_some(1),
None => cfg
.anchor_tokens
.contains(token)
.then_some(2),
}
}
fn can_extend_path<W>(
graph: &StableDiGraph<W>,
state: &CandidatePathState<'_, W>,
next_node: NodeIndex,
target: NodeIndex,
edge: &EdgeData<W>,
cfg: &CandidateSearchConfig<'_>,
) -> bool {
let next_addr = &graph[next_node];
if state
.path
.edge_iter()
.iter()
.any(|existing| existing.component_id == edge.component_id)
{
return false;
}
if state.path.tokens.contains(&next_addr) {
return false;
}
if next_addr == cfg.source_token {
return false;
}
if next_node == target {
return true;
}
cfg.connector_tokens
.map(|tokens| tokens.contains(next_addr))
.unwrap_or(true)
}
fn simulate_edge<W>(
market: &MarketDataView<'_>,
amount: &BigUint,
token_in_addr: &Address,
edge: &EdgeData<W>,
token_out_addr: &Address,
) -> Option<BigUint> {
let token_in = market.get_token(token_in_addr)?;
let token_out = market.get_token(token_out_addr)?;
let state = market.get_simulation_state(&edge.component_id)?;
state
.get_amount_out(amount.clone(), token_in, token_out)
.ok()
.map(|result| result.amount)
}
fn select_candidate_edges<W>(
mut scored: Vec<ScoredEdge<'_, W>>,
max_edges: usize,
) -> Vec<ScoredEdge<'_, W>> {
scored.sort_by(compare_scored_edges);
let mut selected = Vec::new();
let mut per_target: HashMap<NodeIndex, usize> = HashMap::new();
for edge in scored {
let limit = if edge.priority == 0 {
CANDIDATE_DIRECT_EDGES_PER_TOKEN
} else {
CANDIDATE_CONNECTOR_EDGES_PER_TOKEN
};
let count = per_target
.entry(edge.target)
.or_default();
if *count >= limit {
continue;
}
*count += 1;
selected.push(edge);
if selected.len() >= max_edges {
break;
}
}
selected
}
fn compare_scored_edges<W>(a: &ScoredEdge<'_, W>, b: &ScoredEdge<'_, W>) -> Ordering {
a.priority
.cmp(&b.priority)
.then_with(|| b.amount_out.cmp(&a.amount_out))
}
fn prune_candidate_frontier<W>(
by_node: HashMap<NodeIndex, Vec<CandidatePathState<'_, W>>>,
) -> Vec<CandidatePathState<'_, W>> {
by_node
.into_values()
.flat_map(|mut states| {
states.sort_by(|a, b| b.amount_out.cmp(&a.amount_out));
states.truncate(CANDIDATE_STATES_PER_NODE);
states
})
.collect()
}
fn rank_found_candidate_paths<'a, W>(
mut found: Vec<(Path<'a, W>, BigUint)>,
max_candidates: usize,
order: &Order,
) -> Result<CandidatePathSet<'a, W>, AlgorithmError> {
found.sort_by(|(_, a), (_, b)| b.cmp(a));
let mut keys = HashSet::new();
let mut paths = Vec::new();
let mut scores = Vec::new();
for (path, amount_out) in found {
if !keys.insert(path_key(&path)) {
continue;
}
let idx = paths.len();
paths.push(path);
scores.push((idx, BigInt::from(amount_out)));
if paths.len() >= max_candidates {
break;
}
}
if paths.is_empty() {
return Err(AlgorithmError::NoPath {
from: order.token_in().clone(),
to: order.token_out().clone(),
reason: NoPathReason::NoGraphPath,
});
}
Ok((paths, scores))
}
#[cfg(test)]
mod tests {
use std::time::Duration;
use alloy::primitives::U256;
use num_bigint::BigUint;
use num_traits::ToPrimitive;
use tycho_simulation::evm::protocol::uniswap_v2::state::UniswapV2State;
use super::*;
use crate::{
algorithm::{
split_test_harness::{
evaluate_scenario, optimal_two_pool_output, split_metrics, split_scenarios,
two_equal_weth_usdc, TWO_EQUAL_USDC_RESERVE, TWO_EQUAL_WETH_RESERVE,
},
test_utils::{
addr, setup_market_unweighted, setup_market_weighted_boxed, token_with_decimals,
ConstantProductSim,
},
},
graph::GraphManager,
types::quote::OrderSide,
};
fn config() -> AlgorithmConfig {
config_ms(2000)
}
fn config_ms(ms: u64) -> AlgorithmConfig {
AlgorithmConfig::new(1, 3, Duration::from_millis(ms), None).unwrap()
}
fn whole_weth_order(token_in: &Address, token_out: &Address, weth: u64) -> Order {
Order::new(
token_in.clone(),
token_out.clone(),
BigUint::from(weth) * BigUint::from(10u64).pow(18),
OrderSide::Sell,
addr(0xFF),
)
}
fn v2_pool(reserve_a: u128, reserve_b: u128) -> UniswapV2State {
UniswapV2State::new(
U256::from(reserve_a) * U256::from(10u64).pow(U256::from(18u64)),
U256::from(reserve_b) * U256::from(10u64).pow(U256::from(18u64)),
)
}
fn water_fill_default() -> WaterFillAlgorithm {
WaterFillAlgorithm::with_config(
AlgorithmConfig::new(1, 4, Duration::from_millis(5000), None).unwrap(),
)
.unwrap()
}
#[tokio::test]
async fn test_water_fill_all_scenarios() {
let algo = water_fill_default();
for scenario in split_scenarios::all() {
let name = scenario.name;
let (market, gm) = scenario.build_market_weighted();
let result = evaluate_scenario(&algo, &scenario, market, gm).await;
result.assert_passes_lower_bound();
let tolerance_pct = if name == "DOUBLE_SPLIT" { 10 } else { 5 };
assert!(
result.within_pct_of_optimum(tolerance_pct),
"'{name}': not within {tolerance_pct}% of optimum",
);
let route = result
.route
.as_ref()
.unwrap_or_else(|| panic!("'{name}': expected a route"));
assert!(route.validate().is_ok(), "'{name}': route validation failed");
}
}
#[tokio::test]
async fn test_water_fill_gas_kills_split() {
let scenario = split_scenarios::gas_kills_split();
let (market, gm) = scenario.build_market_weighted();
let result = evaluate_scenario(&water_fill_default(), &scenario, market, gm).await;
assert_eq!(result.path_count, 1, "high gas should prevent splitting");
}
#[tokio::test]
async fn test_split_fills_when_no_single_path_can() {
let a = token_with_decimals(0x01, "A", 18);
let b = token_with_decimals(0x02, "B", 18);
let pool = || {
Box::new(ConstantProductSim {
reserve_0: BigUint::from(800u64) * BigUint::from(10u64).pow(18),
reserve_1: BigUint::from(10_000u64) * BigUint::from(10u64).pow(18),
gas: 50_000,
}) as Box<dyn ProtocolSim>
};
let (market, gm) = setup_market_weighted_boxed(vec![
("pool_x", &a, &b, pool()),
("pool_y", &a, &b, pool()),
]);
let order = Order::new(
a.address.clone(),
b.address.clone(),
BigUint::from(1_000u64) * BigUint::from(10u64).pow(18),
OrderSide::Sell,
addr(0xFF),
);
let single = MostLiquidAlgorithm::with_config(config())
.unwrap()
.find_best_route(gm.graph(), market.clone(), None, None, &order)
.await;
assert!(single.is_err(), "premise: no single path fills the full order");
let split = WaterFillAlgorithm::with_config(config())
.unwrap()
.find_best_route(gm.graph(), market.clone(), None, None, &order)
.await
.expect("water_fill returns a split when no single path fills");
assert!(split.route().swaps().len() >= 2, "expected a split across both pools");
}
#[tokio::test]
async fn test_water_fill_output_near_two_pool_optimum() {
let m = two_equal_weth_usdc(1);
let trade = 500u64;
let order = whole_weth_order(&m.weth, &m.usdc, trade);
let result = WaterFillAlgorithm::with_config(config())
.unwrap()
.find_best_route(
m.weighted.graph(),
m.market.clone(),
None,
Some(m.derived.clone()),
&order,
)
.await
.expect("split solves");
let (_, path_count, gross) = split_metrics(&result, &m.weth, &m.usdc);
assert_eq!(path_count, 2, "the optimum uses both pools");
let reserve_in = TWO_EQUAL_WETH_RESERVE as f64 * 1e18;
let reserve_out = TWO_EQUAL_USDC_RESERVE as f64 * 1e6;
let trade_amount = trade as f64 * 1e18;
let (_, optimum) =
optimal_two_pool_output(reserve_in, reserve_out, reserve_in, reserve_out, trade_amount);
let gross = gross.to_f64().unwrap();
assert!(
gross >= optimum * 0.999 && gross <= optimum * 1.0001,
"split gross {gross} should be within 0.1% of the two-pool optimum {optimum}",
);
}
#[tokio::test]
async fn test_water_fill_no_loss_under_tight_timeout() {
for ms in [1u64, 5, 50] {
let m = two_equal_weth_usdc(1_000_000_000);
let order = whole_weth_order(&m.weth, &m.usdc, 500);
let split = WaterFillAlgorithm::with_config(config_ms(ms))
.unwrap()
.find_best_route(
m.weighted.graph(),
m.market.clone(),
None,
Some(m.derived.clone()),
&order,
)
.await;
let single = MostLiquidAlgorithm::with_config(config_ms(ms))
.unwrap()
.find_best_route(
m.weighted.graph(),
m.market.clone(),
None,
Some(m.derived.clone()),
&order,
)
.await;
let (Ok(split), Ok(single)) = (split, single) else {
continue;
};
let (split_net, _, _) = split_metrics(&split, &m.weth, &m.usdc);
let (single_net, _, _) = split_metrics(&single, &m.weth, &m.usdc);
assert!(
split_net >= single_net,
"split lost to single-path under {ms}ms timeout: split={split_net} single={single_net}",
);
}
}
#[tokio::test]
async fn test_discovery_finds_and_ranks_parallel_pools() {
let link = token_with_decimals(0x01, "LINK", 18);
let weth = token_with_decimals(0x02, "WETH", 18);
let (market, graph_manager) = setup_market_unweighted(vec![
(
"a_weak_link_weth",
&link,
&weth,
Box::new(v2_pool(2_000_000, 264)) as Box<dyn ProtocolSim>,
),
(
"z_strong_link_weth",
&link,
&weth,
Box::new(v2_pool(2_000_000, 5_700)) as Box<dyn ProtocolSim>,
),
]);
let order = Order::new(
link.address.clone(),
weth.address.clone(),
BigUint::from(1_000u64) * BigUint::from(10u64).pow(18),
OrderSide::Sell,
addr(0xFF),
);
let start = Instant::now();
let view = market.read().await;
let (paths, scores) = find_candidate_paths(
graph_manager.graph(),
&view,
&order,
CandidateSearchConfig {
min_hops: 1,
max_hops: 3,
max_candidates: 128,
connector_tokens: None,
anchor_tokens: &HashSet::new(),
source_token: order.token_in(),
start: &start,
timeout_ms: 2000,
},
)
.expect("discovery finds candidates");
assert_eq!(paths.len(), 2, "both parallel pools should be discovered");
let best_path = &paths[scores[0].0];
assert_eq!(
best_path.edge_iter()[0].component_id,
"z_strong_link_weth",
"discovery should rank by simulated output, not topology or edge weights",
);
}
#[tokio::test]
async fn test_discovery_rejects_invalid_hop_configuration() {
let link = token_with_decimals(0x01, "LINK", 18);
let weth = token_with_decimals(0x02, "WETH", 18);
let (market, graph_manager) = setup_market_unweighted(vec![(
"link_weth",
&link,
&weth,
Box::new(v2_pool(2_000_000, 5_700)) as Box<dyn ProtocolSim>,
)]);
let order = Order::new(
link.address.clone(),
weth.address.clone(),
BigUint::from(1_000u64) * BigUint::from(10u64).pow(18),
OrderSide::Sell,
addr(0xFF),
);
let start = Instant::now();
let view = market.read().await;
let result = find_candidate_paths(
graph_manager.graph(),
&view,
&order,
CandidateSearchConfig {
min_hops: 0,
max_hops: 3,
max_candidates: 128,
connector_tokens: None,
anchor_tokens: &HashSet::new(),
source_token: order.token_in(),
start: &start,
timeout_ms: 2000,
},
);
assert!(
matches!(result, Err(AlgorithmError::InvalidConfiguration { .. })),
"min_hops of 0 should be rejected",
);
}
#[test]
fn test_derive_anchor_tokens_ranks_hub_and_includes_native_sentinel() {
let hub = token_with_decimals(0x01, "HUB", 18);
let a = token_with_decimals(0x02, "A", 18);
let b = token_with_decimals(0x03, "B", 18);
let c = token_with_decimals(0x04, "C", 18);
let (_market, graph_manager) = setup_market_unweighted(vec![
("hub_a", &hub, &a, Box::new(v2_pool(1, 1)) as Box<dyn ProtocolSim>),
("hub_b", &hub, &b, Box::new(v2_pool(1, 1)) as Box<dyn ProtocolSim>),
("hub_c", &hub, &c, Box::new(v2_pool(1, 1)) as Box<dyn ProtocolSim>),
]);
let anchors = derive_anchor_tokens(graph_manager.graph());
assert!(anchors.contains(&hub.address), "highest-degree token should be anchored");
assert!(
anchors.contains(&Address::from([0u8; 20])),
"native-ETH sentinel should always be anchored",
);
}
}