Robust pricing?

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,9 +402,8 @@ 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<f64> = sqlx::query_scalar(
+        let exact: Option<f64> =
-            "SELECT close FROM ohlcv_1h WHERE token_id = ? AND bucket_ts = ?",
+            sqlx::query_scalar("SELECT close FROM ohlcv_1h WHERE token_id = ? AND bucket_ts = ?")
-        )
                 .bind(token_blob)
                 .bind(scan_start)
                 .fetch_optional(pool)
@@ -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);

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
+    /// Nothing beyond reserves needs priming: credit is derived from them (see
-    /// present depth.  Pools sitting far off-reference start at zero and must earn credit
+    /// [`credit_from`]), so a seeded policy and one that folded its way to the same
-    /// through an accepted print, so a snapshot cannot hand qualification to a pool that
+    /// reserves are indistinguishable by construction.
-    /// was walked away from the market before the window opened.
     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 =
+        let price = leg.sats_delta.unsigned_abs() as f64 / leg.token_delta.unsigned_abs() as f64;
-            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);
+        // Both sides derived from present reserves, so this scan is the whole of tier 2's
-        let largest_credit = self.pools.values().map(|s| s.credit).max().unwrap_or(0);
+        // state. It is O(pools), but only reached by off-band legs — `WithinBand` returns
-        // `ceil().max(1)` keeps a pool that has never had an accepted print (credit 0)
+        // above — so the common path stays a single comparison.
-        // from qualifying just because the largest credit is small.
+        let f = self.params.max_deviation_factor;
-        let threshold = ((largest_credit as f64 * self.params.min_share_fraction).ceil() as u64).max(1);
+        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<f64> {
     if token_amount < MIN_TOKEN_RESERVE || sats == 0 {
@@ -300,7 +326,10 @@ fn spot_ratio(sats: u64, token_amount: u64) -> Option<f64> {
 
 /// 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<f64> {
 /// 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(
+fn median_by(sorted: &[(f64, u64, u64)], weight: impl Fn(&(f64, u64, u64)) -> u64) -> Option<f64> {
-    sorted: &[(f64, u64, u64)],
-    weight: impl Fn(&(f64, u64, u64)) -> u64,
-) -> Option<f64> {
     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.

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));
+    let judged = step(
-    assert!(matches!(judged.verdict, Verdict::Muted { .. }), "{:?}", judged.verdict);
+        &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
+/// reserve 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.
+/// 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);
@@ -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.

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())

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,9 +656,8 @@ mod tests {
             .await
             .unwrap();
 
-        let (close, open, high, low): (f64, f64, f64, f64) = sqlx::query_as(
+        let (close, open, high, low): (f64, f64, f64, f64) =
-            "SELECT close, open, high, low FROM ohlcv_1h WHERE token_id = ?",
+            sqlx::query_as("SELECT close, open, high, low FROM ohlcv_1h WHERE token_id = ?")
-        )
                 .bind(token.as_slice())
                 .fetch_one(&pool)
                 .await