From f0a1b2cf138ce11714ae629c1783516a161774fd Mon Sep 17 00:00:00 2001 From: jakobsn Date: Thu, 30 Jul 2026 11:42:10 +0200 Subject: [PATCH] Robust pricing? --- src/db/cauldron/candlestick/mod.rs | 17 ++- src/db/cauldron/candlestick/policy.rs | 120 ++++++++++++-------- src/db/cauldron/candlestick/policy/tests.rs | 102 ++++++++++++++--- src/db/cauldron/candlestick/tests.rs | 4 +- src/db/cauldron/ohlcv.rs | 25 ++-- 5 files changed, 185 insertions(+), 83 deletions(-) diff --git a/src/db/cauldron/candlestick/mod.rs b/src/db/cauldron/candlestick/mod.rs index d82f5bb..4001c59 100644 --- a/src/db/cauldron/candlestick/mod.rs +++ b/src/db/cauldron/candlestick/mod.rs @@ -134,7 +134,7 @@ fn aggregate_raw_trades( } } - policy.apply(leg, judge.accepted()); + policy.apply(leg); leg_index += 1; } @@ -402,13 +402,12 @@ async fn fetch_last_close_before( // closes on a print that happened *after* the instant being asked about, which an // earlier revision leaked for any non-hour-aligned `timestamp_end`. if scan_start + 3600 <= timestamp_end { - let exact: Option = sqlx::query_scalar( - "SELECT close FROM ohlcv_1h WHERE token_id = ? AND bucket_ts = ?", - ) - .bind(token_blob) - .bind(scan_start) - .fetch_optional(pool) - .await?; + let exact: Option = + sqlx::query_scalar("SELECT close FROM ohlcv_1h WHERE token_id = ? AND bucket_ts = ?") + .bind(token_blob) + .bind(scan_start) + .fetch_optional(pool) + .await?; if exact.is_some() { return Ok(exact); @@ -427,7 +426,7 @@ async fn fetch_last_close_before( if judge.accepted() { last_accepted_price = judge.price; } - policy.apply(leg, judge.accepted()); + policy.apply(leg); } if last_accepted_price.is_some() { return Ok(last_accepted_price); diff --git a/src/db/cauldron/candlestick/policy.rs b/src/db/cauldron/candlestick/policy.rs index 94ca044..a49966c 100644 --- a/src/db/cauldron/candlestick/policy.rs +++ b/src/db/cauldron/candlestick/policy.rs @@ -106,14 +106,16 @@ impl Default for GuardParams { } } -/// Per-pool state: reserves and qualification credit. +/// Per-pool state: reserves only. +/// +/// Qualification credit is *derived* from these on demand (see [`credit_from`]) rather than +/// stored, which is what keeps a policy's verdicts a function of the reserves it holds +/// rather than of the order it learned them in. #[derive(Clone, Debug)] struct PoolState { sequence: i64, sats: u64, token_amount: u64, - /// Min-depth at this pool's last accepted print. - credit: u64, } /// The guard: judges each leg against a reference price derived from pool reserves. @@ -140,10 +142,9 @@ impl Policy { /// A policy primed with pool reserves as of some instant. /// - /// Credit is seeded for pools already trading within `F` of the reference, capped by - /// present depth. Pools sitting far off-reference start at zero and must earn credit - /// through an accepted print, so a snapshot cannot hand qualification to a pool that - /// was walked away from the market before the window opened. + /// Nothing beyond reserves needs priming: credit is derived from them (see + /// [`credit_from`]), so a seeded policy and one that folded its way to the same + /// reserves are indistinguishable by construction. pub fn seeded(params: GuardParams, snapshot: &[PoolReserves]) -> Self { let mut policy = Self { pools: HashMap::with_capacity(snapshot.len()), @@ -159,24 +160,10 @@ impl Policy { sequence: r.sequence, sats: r.sats, token_amount: r.token_amount, - credit: 0, }, ); } policy.recompute_reference(); - - if let Some(reference) = policy.reference { - let f = policy.params.max_deviation_factor; - for state in policy.pools.values_mut() { - let Some(spot) = spot_ratio(state.sats, state.token_amount) else { - continue; - }; - if deviation(spot, reference) <= f { - state.credit = min_depth(state.sats, state.token_amount, reference); - } - } - } - policy } @@ -190,8 +177,7 @@ impl Policy { }; } - let price = - leg.sats_delta.unsigned_abs() as f64 / leg.token_delta.unsigned_abs() as f64; + let price = leg.sats_delta.unsigned_abs() as f64 / leg.token_delta.unsigned_abs() as f64; let Some(reference) = self.reference else { return JudgeResult { @@ -208,11 +194,24 @@ impl Policy { }; } - let credit = self.pools.get(&leg.pool).map_or(0, |s| s.credit); - let largest_credit = self.pools.values().map(|s| s.credit).max().unwrap_or(0); - // `ceil().max(1)` keeps a pool that has never had an accepted print (credit 0) - // from qualifying just because the largest credit is small. - let threshold = ((largest_credit as f64 * self.params.min_share_fraction).ceil() as u64).max(1); + // Both sides derived from present reserves, so this scan is the whole of tier 2's + // state. It is O(pools), but only reached by off-band legs — `WithinBand` returns + // above — so the common path stays a single comparison. + let f = self.params.max_deviation_factor; + let credit = self + .pools + .get(&leg.pool) + .map_or(0, |s| credit_from(s.sats, s.token_amount, reference, f)); + let largest_credit = self + .pools + .values() + .map(|s| credit_from(s.sats, s.token_amount, reference, f)) + .max() + .unwrap_or(0); + // `ceil().max(1)` keeps a pool with no usable depth (credit 0) from qualifying just + // because the largest credit is small. + let threshold = + ((largest_credit as f64 * self.params.min_share_fraction).ceil() as u64).max(1); let verdict = if credit >= threshold { Verdict::CreditExempt { @@ -236,15 +235,14 @@ impl Policy { /// Advance state past a leg. Must be called for every leg, accepted or not: a muted /// leg still moved real reserves. - pub fn apply(&mut self, leg: &Leg, accepted: bool) { - let reference = self.reference; - + /// + /// Takes no verdict, because none is needed — credit follows the reserves this records. + pub fn apply(&mut self, leg: &Leg) { // `i64::MIN` so a pool's very first leg clears the monotonicity guard below. let state = self.pools.entry(leg.pool).or_insert(PoolState { sequence: i64::MIN, sats: 0, token_amount: 0, - credit: 0, }); // Strict `>`: the `i64::MIN` sentinel above is what lets a pool's first leg through, @@ -253,15 +251,6 @@ impl Policy { state.sequence = leg.sequence; state.sats = leg.sats; state.token_amount = leg.token_amount; - - if accepted && leg.token_delta != 0 { - // Value the token side at the reference the leg was judged against; for a - // token's very first print there is none, so use the leg's own price. - let valuation = reference.unwrap_or_else(|| { - leg.sats_delta.unsigned_abs() as f64 / leg.token_delta.unsigned_abs() as f64 - }); - state.credit = min_depth(leg.sats, leg.token_amount, valuation); - } } self.recompute_reference(); @@ -275,7 +264,17 @@ impl Policy { /// Qualification credit held by a pool. Exposed for tests and diagnostics. #[allow(dead_code)] pub fn credit_of(&self, pool: &[u8; 32]) -> u64 { - self.pools.get(pool).map_or(0, |s| s.credit) + let Some(reference) = self.reference else { + return 0; + }; + self.pools.get(pool).map_or(0, |s| { + credit_from( + s.sats, + s.token_amount, + reference, + self.params.max_deviation_factor, + ) + }) } fn recompute_reference(&mut self) { @@ -290,6 +289,33 @@ impl Policy { } } +/// Qualification credit: a pool's min-depth, but only while its own spot sits within `F` of +/// the reference. +/// +/// Stateless by construction, and that is the point. An earlier revision stored "min-depth +/// at the pool's last accepted print", which made credit path-dependent — a reserve +/// snapshot cannot recover it, so [`Policy::seeded`] had to approximate, and a seeded policy +/// returned `CreditExempt` where a folded one returned `Muted` on the very same leg. That +/// broke the invariant the whole seeding design exists to hold, and because `ohlcv_1h` is +/// written with `INSERT OR IGNORE`, whichever path ran first would freeze its answer in +/// permanently. +/// +/// Little security was given up. The path-dependent rule was a speed bump rather than a +/// wall: an attacker needed one cheap in-band trade to convert a fresh pool into a credited +/// one. What remains is a pure depth gate — printing far off-reference requires a pool +/// holding a real share of the token's liquidity, which is the part that actually costs an +/// attacker money. A pool that has been walked off-market is still excluded outright, so a +/// snapshot cannot launder an attacker's pool into qualification. +fn credit_from(sats: u64, token_amount: u64, reference: f64, f: f64) -> u64 { + let Some(spot) = spot_ratio(sats, token_amount) else { + return 0; + }; + if deviation(spot, reference) > f { + return 0; + } + min_depth(sats, token_amount, reference) +} + /// A pool's spot price, or `None` if it is too small or one-sided to quote one. fn spot_ratio(sats: u64, token_amount: u64) -> Option { if token_amount < MIN_TOKEN_RESERVE || sats == 0 { @@ -300,7 +326,10 @@ fn spot_ratio(sats: u64, token_amount: u64) -> Option { /// How far apart two prices are, as a factor >= 1 in whichever direction. fn deviation(price: f64, reference: f64) -> f64 { - if !(price > 0.0) || !(reference > 0.0) || !price.is_finite() || !reference.is_finite() { + // Finiteness is tested first so the `<=` comparisons below are total: NaN fails + // `is_finite` and returns here, which is why plain `<=` is safe rather than needing + // `!(price > 0.0)` to catch it. + if !price.is_finite() || !reference.is_finite() || price <= 0.0 || reference <= 0.0 { return f64::INFINITY; } (price / reference).max(reference / price) @@ -360,10 +389,7 @@ fn weighted_median_reference(entries: &mut [(f64, u64, u64)]) -> Option { /// Compares `2 * cumulative >= total` rather than `cumulative >= total / 2` so integer /// division cannot bias the pick downwards. When every weight is zero it falls back to /// the positional median, so an all-dust set does not collapse to its smallest ratio. -fn median_by( - sorted: &[(f64, u64, u64)], - weight: impl Fn(&(f64, u64, u64)) -> u64, -) -> Option { +fn median_by(sorted: &[(f64, u64, u64)], weight: impl Fn(&(f64, u64, u64)) -> u64) -> Option { let total: u128 = sorted.iter().map(|e| weight(e) as u128).sum(); if total == 0 { // Defensive only: `spot_ratio` rejects `sats == 0`, so every live weight is >= 1. diff --git a/src/db/cauldron/candlestick/policy/tests.rs b/src/db/cauldron/candlestick/policy/tests.rs index bd8b98f..b5662d5 100644 --- a/src/db/cauldron/candlestick/policy/tests.rs +++ b/src/db/cauldron/candlestick/policy/tests.rs @@ -45,7 +45,7 @@ fn reserves(pool: u8, sequence: i64, sats: u64, token_amount: u64) -> PoolReserv /// Feed a leg through judge-then-apply, returning the verdict. fn step(policy: &mut Policy, l: &Leg) -> JudgeResult { let judged = policy.judge(l); - policy.apply(l, judged.accepted()); + policy.apply(l); judged } @@ -95,18 +95,26 @@ fn test_leg_far_from_the_reference_without_credit_is_muted() { step(&mut policy, &leg(1, 1, 100_000, -1_000, 1_000_000, 10_000)); // A fresh pool printing 100x off market, with no accepted print behind it. - let judged = step(&mut policy, &leg(9, 2, 10_000_000, -1_000, 10_001_000, 1_010)); - assert!(matches!(judged.verdict, Verdict::Muted { .. }), "{:?}", judged.verdict); + let judged = step( + &mut policy, + &leg(9, 2, 10_000_000, -1_000, 10_001_000, 1_010), + ); + assert!( + matches!(judged.verdict, Verdict::Muted { .. }), + "{:?}", + judged.verdict + ); assert!(!judged.accepted()); // Muted, but the price was still computed — the caller still counts its volume. assert!(judged.price.is_some()); } /// Regression: `apply` inserted new pools with `sequence: leg.sequence` and then gated the -/// credit update behind `leg.sequence > state.sequence`, which is false on that very first -/// insert. A pool therefore had to make two accepted prints before earning any credit. +/// reserve update behind `leg.sequence > state.sequence`, which is false on that very first +/// insert — so a pool's opening leg left it recorded with zero reserves, and therefore zero +/// credit. The `i64::MIN` sentinel is what fixes it. #[test] -fn test_pool_earns_credit_on_its_first_accepted_print() { +fn test_pool_holds_credit_after_its_very_first_leg() { let mut policy = Policy::new(params()); step(&mut policy, &leg(1, 1, 100_000, -1_000, 1_000_000, 10_000)); @@ -116,23 +124,68 @@ fn test_pool_earns_credit_on_its_first_accepted_print() { assert!( policy.credit_of(&[2; 32]) > 0, - "a pool must hold credit after its first accepted print, not its second" + "a pool's opening leg must record its reserves, not leave it at zero" ); } +/// A wild print leaves the pool that made it sitting off-market, and an off-market pool +/// holds no credit — so the pool cannot use one absurd leg to qualify the next. #[test] -fn test_muted_leg_does_not_earn_credit() { +fn test_pool_left_off_market_by_its_own_print_holds_no_credit() { let mut policy = Policy::new(params()); step(&mut policy, &leg(1, 1, 100_000, -1_000, 1_000_000, 10_000)); - step(&mut policy, &leg(9, 2, 10_000_000, -1_000, 10_001_000, 1_010)); + step( + &mut policy, + &leg(9, 2, 10_000_000, -1_000, 10_001_000, 1_010), + ); assert_eq!( policy.credit_of(&[9; 32]), 0, - "a muted print must not qualify the pool that made it" + "a pool 99x off the reference must not qualify for tier 2" ); } +/// Regression for the seeded/folded credit divergence — the reason credit is derived from +/// reserves rather than remembered. +/// +/// Pool 2's only leg is a one-token trade at 100x the market. It is muted, but it is far too +/// small to move the pool's own spot, so the pool is left deep and in-band. Under the old +/// "min-depth at the last accepted print" rule the folded policy held pool 2 at zero credit +/// while a policy seeded from the resulting reserves granted it full credit — and the two +/// then returned `Muted` and `CreditExempt` for the identical next leg. Since `ohlcv_1h` is +/// written with `INSERT OR IGNORE`, whichever path ran first would have frozen its answer in. +#[test] +fn test_seeded_and_folded_agree_on_credit_after_a_muted_print() { + let mut folded = Policy::new(params()); + step(&mut folded, &leg(1, 1, 100_000, -1_000, 1_000_000, 10_000)); + // Pool 2's opening leg: absurd price, negligible size, so it ends in-band. + step(&mut folded, &leg(2, 2, 10_000, -1, 2_010_000, 19_999)); + + let seeded = Policy::seeded( + params(), + &[ + reserves(1, 1, 1_000_000, 10_000), + reserves(2, 2, 2_010_000, 19_999), + ], + ); + + assert_eq!(folded.reference(), seeded.reference()); + assert_eq!( + folded.credit_of(&[2; 32]), + seeded.credit_of(&[2; 32]), + "credit must not depend on whether the policy folded history or was seeded from it" + ); + assert!( + folded.credit_of(&[2; 32]) > 0, + "the test is vacuous unless the pool actually ends up in-band and credited" + ); + + // The consequence that made this worth fixing: identical verdicts on the next leg. + let next = leg(2, 3, 500_000, -1, 2_510_000, 19_998); + assert_eq!(folded.judge(&next).verdict, seeded.judge(&next).verdict); +} + /// The property the whole seeded design exists for: the reference depends only on pool /// reserves, never on how many legs the caller happened to read to arrive at them. #[test] @@ -145,7 +198,10 @@ fn test_reference_is_a_function_of_reserves_not_of_history_read() { // Same end state, reached by seeding instead of folding. let seeded = Policy::seeded( params(), - &[reserves(1, 3, 1_105_000, 9_000), reserves(2, 2, 2_000_000, 20_000)], + &[ + reserves(1, 3, 1_105_000, 9_000), + reserves(2, 2, 2_000_000, 20_000), + ], ); assert_eq!( @@ -165,7 +221,10 @@ fn test_seeded_and_folded_policies_agree_on_a_shared_leg() { let seeded = Policy::seeded( params(), - &[reserves(1, 1, 1_000_000, 10_000), reserves(2, 2, 2_000_000, 20_000)], + &[ + reserves(1, 1, 1_000_000, 10_000), + reserves(2, 2, 2_000_000, 20_000), + ], ); let shared = leg(1, 3, 300_000, -1_000, 1_300_000, 9_000); @@ -184,9 +243,9 @@ fn test_seeding_withholds_credit_from_off_market_pools() { let policy = Policy::seeded( params(), &[ - reserves(1, 1, 1_000_000, 10_000), // spot 100, at market - reserves(2, 2, 2_000_000, 20_000), // spot 100, at market - reserves(9, 3, 10_000_000, 1_000), // spot 10_000, 100x off + reserves(1, 1, 1_000_000, 10_000), // spot 100, at market + reserves(2, 2, 2_000_000, 20_000), // spot 100, at market + reserves(9, 3, 10_000_000, 1_000), // spot 10_000, 100x off ], ); @@ -238,7 +297,9 @@ fn test_pools_that_cannot_quote_a_ratio_are_excluded() { ], ); - let reference = policy.reference().expect("the one quotable pool sets the reference"); + let reference = policy + .reference() + .expect("the one quotable pool sets the reference"); assert!( reference.is_finite() && reference == 100.0, "got {reference}" @@ -356,6 +417,15 @@ fn test_deviation_is_symmetric_and_rejects_degenerate_input() { assert_eq!(deviation(50.0, 100.0), 2.0); assert_eq!(deviation(0.0, 100.0), f64::INFINITY); assert_eq!(deviation(100.0, 0.0), f64::INFINITY); + // NaN and the infinities must land on INFINITY (i.e. "out of band"), never sneak + // through as a small deviation. `<=` alone is false for NaN, so the finiteness test + // has to run first — this is what pins that ordering. + assert_eq!(deviation(f64::NAN, 100.0), f64::INFINITY); + assert_eq!(deviation(100.0, f64::NAN), f64::INFINITY); + assert_eq!(deviation(f64::INFINITY, 100.0), f64::INFINITY); + assert_eq!(deviation(100.0, f64::INFINITY), f64::INFINITY); + assert_eq!(deviation(-100.0, 100.0), f64::INFINITY); + assert_eq!(deviation(100.0, -100.0), f64::INFINITY); } /// A leg replayed at or below a pool's known sequence must not rewind its reserves. diff --git a/src/db/cauldron/candlestick/tests.rs b/src/db/cauldron/candlestick/tests.rs index e325cc4..ffff7b8 100644 --- a/src/db/cauldron/candlestick/tests.rs +++ b/src/db/cauldron/candlestick/tests.rs @@ -69,7 +69,9 @@ async fn register_pool( .unwrap(); if let Some((spent, txid, ts)) = withdrawn_utxo { - insert_mempool_tx(&mut **conn, &txid, ts as u64).await.unwrap(); + insert_mempool_tx(&mut **conn, &txid, ts as u64) + .await + .unwrap(); sqlx::query("INSERT OR REPLACE INTO utxo_spending (spent_utxo_hash, txid) VALUES (?, ?)") .bind(spent.to_blob()) .bind(txid.to_blob()) diff --git a/src/db/cauldron/ohlcv.rs b/src/db/cauldron/ohlcv.rs index 01892fa..95b6f67 100644 --- a/src/db/cauldron/ohlcv.rs +++ b/src/db/cauldron/ohlcv.rs @@ -22,7 +22,10 @@ use sqlx::{Row, SqlitePool}; /// written flat at the carried close instead of dropped, restoring their volume; the /// rebuild seeds from confirmed-only reserves, so output no longer depends on mempool /// contents at rebuild time. -pub const OHLCV_VERSION: u32 = 5; +/// 6: qualification credit derived from present reserves instead of remembered from a +/// pool's last accepted print, so a seeded policy and a folded one can no longer +/// disagree on the same leg. Tier 2 is now a pure depth gate. +pub const OHLCV_VERSION: u32 = 6; const OHLCV_VERSION_KEY: &str = "ohlcv_version"; pub async fn create_table(pool: &SqlitePool) { @@ -213,7 +216,7 @@ ORDER BY phe.token_id ASC, phe.effective_timestamp ASC, phe.sequence ASC; } } - policy.apply(&leg, judge.accepted()); + policy.apply(&leg); } // Phase 3: insert all computed buckets in one transaction. @@ -607,7 +610,10 @@ mod tests { .await .unwrap(); - assert_eq!(volume, 25, "volume must survive a bucket with no accepted price"); + assert_eq!( + volume, 25, + "volume must survive a bucket with no accepted price" + ); assert_eq!( (open, close, high, low), (40.0, 40.0, 40.0, 40.0), @@ -650,13 +656,12 @@ mod tests { .await .unwrap(); - let (close, open, high, low): (f64, f64, f64, f64) = sqlx::query_as( - "SELECT close, open, high, low FROM ohlcv_1h WHERE token_id = ?", - ) - .bind(token.as_slice()) - .fetch_one(&pool) - .await - .unwrap(); + let (close, open, high, low): (f64, f64, f64, f64) = + sqlx::query_as("SELECT close, open, high, low FROM ohlcv_1h WHERE token_id = ?") + .bind(token.as_slice()) + .fetch_one(&pool) + .await + .unwrap(); // Per-leg pricing: each leg executes at its own price. // Leg 1 (buy): 446,491,239 / 1,334,527,069 = 0.334569