Fix regressions introduced by the Stage 2 per-leg pricing phases

Review of today's Phase 1-4 commits found six defects, all introduced by those
phases and verified against 283c0c3. Each fix carries a regression test that was
confirmed to fail against the old code.

Reference is now derived from reserves, not folded from an arbitrary start
    Policy was a stateful fold instantiated in three places with three different
    lifetimes: per API request, per 24h rebuild batch, and per seed backscan. A
    fold whose result depends on where you began reading is not a function of the
    chain, which produced three symptoms: rebuild_range reset the policy at every
    batch boundary (~1125 times per token over full history, each reset waving the
    first leg of that day through unjudged via "no reference yet"); candlesticks()
    built a virgin policy per request, so the same hour rendered differently at 1W
    and 1M; and the fast path handed the raw tail a policy that had never seen the
    legs behind the ohlcv_1h buckets preceding it.

    Policy::seeded() now primes reserves from a snapshot query, so every caller
    starts from the same chain state. The d_i = min(S_i, T_i * R) circularity is
    resolved by seeding the weighted median with the unweighted one and reweighting
    to a fixed point, rather than by carrying the previous leg's R forward.

f64::MIN/MAX no longer reach the database
    A bucket whose legs were all muted or unpriceable was inserted with its high/low
    accumulators still at their sentinels, putting +/-1.8e308 into ohlcv_1h and from
    there onto the chart. The SQL this replaced dropped such buckets via an inner
    join; restored that behaviour.

Credit off-by-one
    apply() inserted new pools with sequence: leg.sequence, then gated the credit
    update behind leg.sequence > state.sequence -- false on that very insert. Pools
    needed two accepted prints to earn any credit, weakening tier-2 qualification.

Seed lookback restored to unbounded
    Phase 4 capped the backscan at 24h; the pre-Stage-2 query had no horizon. Tokens
    trading less often than daily lost their carry-forward price entirely.

Hot path and edge cases
    Dropped the per-leg format! allocation in favour of a Copy verdict enum; reused
    a scratch buffer across reference recomputation; fixed the weighted median
    biasing low on integer division and collapsing to the smallest ratio when all
    weights are zero; replaced copy_from_slice with a checked conversion so a
    malformed blob errors instead of panicking. Legs moving tokens for zero sats are
    now unpriceable rather than printing 0.0.

Tests
    test_multipool_arb_priced_by_gross_volume_not_net and
    test_single_direction_multileg_price_matches_net_ratio had their assertions
    rewritten during Phase 4 to match whatever the code produced, leaving names that
    contradicted what they checked. Renamed and rewritten to assert the durable
    invariant: a printed price must be one a leg actually executed at.
    test_credit_seeding_on_accepted_print asserted is_some() on a struct that always
    exists and passed with the credit bug fully present; replaced.

Still open (unbuilt plan phases, not regressions): withdrawal-event synthesis so
withdrawn pools stop voting in the median, credit seeding at pool creation, tier-2
summing credit across a tx's swap legs, swap vs liquidity-event distinction, and
config threading of GuardParams.

OHLCV_VERSION 3 -> 4: version 3 buckets were written by the buggy code and
INSERT OR IGNORE never corrects rows in place.

Tests: 238 passing

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

src/db/cauldron/candlestick/mod.rs

@@ -10,7 +10,15 @@ use bitcoincash::TokenID;
 use serde::Serialize;
 use sqlx::{Row, SqlitePool};
 
-pub use self::policy::{GuardParams, Leg, Policy};
+pub use self::policy::{GuardParams, Leg, Policy, PoolReserves};
+
+/// Interpret a database blob as a 32-byte hash, erroring rather than panicking on a
+/// short or oversized value.
+pub(crate) fn to_hash32(bytes: &[u8]) -> Result<[u8; 32]> {
+    bytes
+        .try_into()
+        .map_err(|_| anyhow::anyhow!("expected a 32-byte hash, got {} bytes", bytes.len()))
+}
 
 #[derive(Debug, Serialize)]
 pub struct CandlestickData {
@@ -102,7 +110,7 @@ fn aggregate_raw_trades(
 
             let judge = policy.judge(leg);
             let sats_delta_abs = leg.sats_delta.unsigned_abs() as i64;
-            let token_delta_abs = leg.token_delta.abs() as i64;
+            let token_delta_abs = leg.token_delta.unsigned_abs() as i64;
 
             // Always count volume, regardless of acceptance.
             pi.volume_sats += sats_delta_abs;
@@ -110,7 +118,7 @@ fn aggregate_raw_trades(
             txid_set.insert(leg.txid);
 
             // Update OHLC only if accepted and priceable.
-            if judge.accepted {
+            if judge.accepted() {
                 if let Some(price) = judge.price {
                     if first_trade_in_interval {
                         pi.open = Some(price);
@@ -126,7 +134,7 @@ fn aggregate_raw_trades(
                 }
             }
 
-            policy.apply(leg, judge.accepted);
+            policy.apply(leg, judge.accepted());
             leg_index += 1;
         }
 
@@ -254,82 +262,134 @@ ORDER BY phe.effective_timestamp ASC, phe.sequence ASC;
         .fetch_all(pool)
         .await?;
 
-    Ok(rows
+    rows.into_iter()
-        .into_iter()
         .map(|r| {
             let txid: Vec<u8> = r.get(0);
             let pool_bytes: Vec<u8> = r.get(2);
-            let mut txid_arr = [0u8; 32];
+            Ok(Leg {
-            let mut pool_arr = [0u8; 32];
+                txid: to_hash32(&txid)?,
-            txid_arr.copy_from_slice(&txid);
+                pool: to_hash32(&pool_bytes)?,
-            pool_arr.copy_from_slice(&pool_bytes);
-            Leg {
-                txid: txid_arr,
-                pool: pool_arr,
                 ts: r.get(1),
                 sequence: r.get(7),
                 sats_delta: r.get(3),
                 token_delta: r.get(4),
                 sats: r.get(5),
                 token_amount: r.get(6),
-            }
             })
-        .collect())
+        })
+        .collect()
 }
 
-/// Returns the close price of the most recent accepted print strictly before
+/// Pool reserves for `token_blob` as they stood immediately before `timestamp`.
-/// `timestamp_end`. Uses ohlcv_1h for fast lookup when available, otherwise
+///
-/// bacscans recent legs through the policy.
+/// Seeding a [`Policy`] with this is what makes its verdicts independent of the caller's
-/// Returns `None` when no prior accepted print exists (new token, no history).
+/// window: the raw path, a rebuild batch, and a 1-week versus 1-month request all start
+/// from the same chain state rather than from whichever legs happened to be in range.
+///
+/// Enumerates the token's pools and seeks each one's latest history row, rather than
+/// scanning every row the token ever produced — a busy token has hundreds of thousands,
+/// and this runs once per query.
+pub(crate) async fn fetch_reserve_snapshot(
+    pool: &SqlitePool,
+    token_blob: &[u8],
+    timestamp: i64,
+) -> Result<Vec<PoolReserves>> {
+    let sql = r#"
+SELECT phe.pool, phe.sequence, phe.sats, phe.token_amount
+FROM pool AS p
+JOIN pool_history_entry AS phe ON phe.utxo = (
+  SELECT prior.utxo
+  FROM pool_history_entry AS prior
+  WHERE prior.pool = p.creation_utxo
+    AND prior.effective_timestamp < ?
+  ORDER BY prior.effective_timestamp DESC, prior.sequence DESC
+  LIMIT 1
+)
+WHERE p.token_id = ?;
+"#;
+    let rows = sqlx::query(sql)
+        .bind(timestamp)
+        .bind(token_blob)
+        .fetch_all(pool)
+        .await?;
+
+    rows.into_iter()
+        .map(|r| {
+            let pool_bytes: Vec<u8> = r.get(0);
+            Ok(PoolReserves {
+                pool: to_hash32(&pool_bytes)?,
+                sequence: r.get(1),
+                sats: r.get(2),
+                token_amount: r.get(3),
+            })
+        })
+        .collect()
+}
+
+/// A price to carry into a window that opens on silence, so that requesting 1W and 1M
+/// yields the same candles over the period they share.
+///
+/// Prefers the most recent accepted print strictly before `timestamp_end`. Falls back to
+/// the reference implied by pool reserves when the token's last activity was entirely
+/// muted or unpriceable — reserves still say what the market was, and a carried price
+/// beats a hole in the chart.
+///
+/// Returns `None` only when the token has no history at all before `timestamp_end`.
+///
+/// The lookback is deliberately unbounded. An earlier revision capped the scan at 24
+/// hours, which silently dropped the seed for any token trading less often than daily.
 async fn fetch_last_close_before(
     pool: &SqlitePool,
     token_blob: &[u8],
     timestamp_end: i64,
 ) -> Result<Option<f64>> {
-    // Fast path: if timestamp_end is hour-aligned, look up the previous bucket's close
+    // Materialised buckets answer this with one index seek on (token_id, bucket_ts).
-    // from the materialized ohlcv_1h table (indexed point lookup).
+    let materialised: Option<f64> = sqlx::query_scalar(
-    if timestamp_end % 3600 == 0 && timestamp_end >= 3600 {
+        "SELECT close FROM ohlcv_1h
-        let prev_bucket = timestamp_end - 3600;
+         WHERE token_id = ? AND bucket_ts < ?
-        let close: Option<f64> = sqlx::query_scalar(
+         ORDER BY bucket_ts DESC LIMIT 1",
-            "SELECT close FROM ohlcv_1h WHERE token_id = ? AND bucket_ts = ?",
     )
     .bind(token_blob)
-        .bind(prev_bucket)
+    .bind(timestamp_end)
     .fetch_optional(pool)
     .await?;
 
-        if close.is_some() {
+    if materialised.is_some() {
-            return Ok(close);
+        return Ok(materialised);
-        }
     }
 
-    // Slow path: if the materialized lookup missed (non-aligned or no bucket), backscan
+    // Nothing materialised: replay the token's most recent active hour through the guard.
-    // recent legs through the policy to find the last accepted print.
+    // Only that hour is needed, because the policy is seeded from reserves rather than
-    // Fetch the last ~500 legs before timestamp_end and fold through them.
+    // rebuilt by folding all of history.
-    let legs = fetch_raw_legs(pool, token_blob, (timestamp_end - 24 * 3600).max(0), timestamp_end)
+    let last_activity: Option<i64> = sqlx::query_scalar(
-        .await?;
+        "SELECT MAX(effective_timestamp) FROM pool_history_entry
+         WHERE token_id = ? AND effective_timestamp < ?",
+    )
+    .bind(token_blob)
+    .bind(timestamp_end)
+    .fetch_optional(pool)
+    .await?
+    .flatten();
 
-    if legs.is_empty() {
+    let Some(last_activity) = last_activity else {
         return Ok(None);
-    }
-
-    let params = GuardParams {
-        max_deviation_factor: 5.0,
-        min_share_fraction: 0.05,
     };
-    let mut policy = Policy::new(params);
 
-    // Fold through legs in chronological order, tracking the last accepted price.
+    let scan_start = (last_activity / 3600) * 3600;
-    let mut last_accepted_price: Option<f64> = None;
+    let snapshot = fetch_reserve_snapshot(pool, token_blob, scan_start).await?;
-    for leg in legs {
+    let mut policy = Policy::seeded(GuardParams::default(), &snapshot);
-        let judge = policy.judge(&leg);
+
-        if judge.accepted {
+    let legs = fetch_raw_legs(pool, token_blob, scan_start, timestamp_end).await?;
+    let mut last_accepted_price = None;
+    for leg in &legs {
+        let judge = policy.judge(leg);
+        if judge.accepted() {
             last_accepted_price = judge.price;
         }
-        policy.apply(&leg, judge.accepted);
+        policy.apply(leg, judge.accepted());
     }
 
-    Ok(last_accepted_price)
+    Ok(last_accepted_price.or_else(|| policy.reference()))
 }
 
 /// `ohlcv_materialized_end`: exclusive upper bound of what is in `ohlcv_1h`.
@@ -347,13 +407,7 @@ pub async fn candlesticks(
     }
 
     let token_blob = display_hex_to_blob::<TokenID>(token_id)?;
-
+    let params = GuardParams::default();
-    // Default guard parameters (F=5, q=5%). TODO: thread from config when flag enabled.
-    let params = GuardParams {
-        max_deviation_factor: 5.0,
-        min_share_fraction: 0.05,
-    };
-    let mut policy = Policy::new(params);
 
     // Seed gap-fill with the last known close price before this window so that
     // switching between timeframes (e.g. 1W vs 1M) produces consistent prices
@@ -383,6 +437,13 @@ pub async fn candlesticks(
 
         if ohlcv_end < timestamp_end {
             // Tail: query raw for [ohlcv_end, timestamp_end) and append.
+            //
+            // The policy is seeded at `ohlcv_end`, not at `timestamp_start`: the buckets
+            // before it came from `ohlcv_1h` and never passed through this policy, so a
+            // policy built from the window's start would enter the tail believing the
+            // token had no history and wave its first leg through unjudged.
+            let snapshot = fetch_reserve_snapshot(pool, &token_blob, ohlcv_end).await?;
+            let mut policy = Policy::seeded(params, &snapshot);
             let raw_legs = fetch_raw_legs(pool, &token_blob, ohlcv_end, timestamp_end).await?;
 
             let mut tail_intervals = Vec::new();
@@ -414,10 +475,18 @@ pub async fn candlesticks(
         current_start += step_size;
     }
 
+    let snapshot = fetch_reserve_snapshot(pool, &token_blob, timestamp_start).await?;
+    let mut policy = Policy::seeded(params, &snapshot);
     let all_legs = fetch_raw_legs(pool, &token_blob, timestamp_start, timestamp_end).await?;
 
-    let (result, _, _) =
+    let (result, _, _) = aggregate_raw_trades(
-        aggregate_raw_trades(&all_legs, intervals, step_size, seed_found, seed_close, &mut policy);
+        &all_legs,
+        intervals,
+        step_size,
+        seed_found,
+        seed_close,
+        &mut policy,
+    );
     Ok(result)
 }
 

src/db/cauldron/candlestick/policy.rs

@@ -3,8 +3,16 @@
 // This software is licensed under the GNU Affero General Public License (AGPL), version 3.0 or later.
 // A copy of the license can be found in the LICENSE file or at https://www.gnu.org/licenses/agpl-3.0.html
 
+use std::cmp::Ordering;
 use std::collections::HashMap;
 
+/// Pools holding fewer tokens than this are dust and do not vote on the reference.
+const MIN_TOKEN_RESERVE: u64 = 10;
+
+/// Upper bound on reweighting passes in [`weighted_median_reference`].  Four is far more
+/// than the one or two the fixed point needs in practice; it exists to bound the loop.
+const REWEIGHT_PASSES: usize = 4;
+
 /// A leg of a transaction: one pool's change in a single transaction.
 #[derive(Clone, Debug)]
 pub struct Leg {
@@ -18,332 +26,352 @@ pub struct Leg {
     pub token_amount: u64, // post-change token_amount (reserves)
 }
 
-/// Outcome of judging a leg for acceptance.
+/// One pool's reserves at an instant, used to seed a [`Policy`].
+///
+/// Seeding is what makes the policy's verdicts a function of the chain rather than of
+/// where the caller happened to start reading: a request for one hour and a request for
+/// one month both begin from the same reserve state and so agree on the overlap.
 #[derive(Clone, Debug)]
-pub struct JudgeResult {
+pub struct PoolReserves {
-    pub price: Option<f64>,  // None if token_delta == 0
+    pub pool: [u8; 32],
-    pub accepted: bool,      // whether this leg is accepted into OHLC
+    pub sequence: i64,
-    pub reason: String,      // debug: why accepted or muted
+    pub sats: u64,
+    pub token_amount: u64,
 }
 
-/// Per-pool state: reserves and credit.
+/// Why a leg was accepted or muted.
+///
+/// Deliberately allocation-free: `judge` runs once per leg, and a full `ohlcv_1h` rebuild
+/// walks millions of them.
+#[derive(Clone, Copy, Debug, PartialEq)]
+pub enum Verdict {
+    /// No token moved, or no sats moved, so the leg has no price.
+    Unpriceable,
+    /// No pool held usable reserves, so there is nothing to deviate from.
+    NoReference,
+    /// Within `max_deviation_factor` of the reference.
+    WithinBand { dev: f64 },
+    /// Outside the band, but the pool holds enough qualification credit to print anyway.
+    CreditExempt {
+        dev: f64,
+        credit: u64,
+        largest_credit: u64,
+    },
+    /// Outside the band with too little credit: counted in volume, excluded from OHLC.
+    Muted {
+        dev: f64,
+        credit: u64,
+        largest_credit: u64,
+    },
+}
+
+impl Verdict {
+    pub fn accepted(self) -> bool {
+        matches!(
+            self,
+            Verdict::NoReference | Verdict::WithinBand { .. } | Verdict::CreditExempt { .. }
+        )
+    }
+}
+
+/// Outcome of judging a leg.
+#[derive(Clone, Copy, Debug)]
+pub struct JudgeResult {
+    /// The leg's executed price, or `None` when it has none.
+    pub price: Option<f64>,
+    pub verdict: Verdict,
+}
+
+impl JudgeResult {
+    pub fn accepted(&self) -> bool {
+        self.verdict.accepted()
+    }
+}
+
+#[derive(Clone, Copy, Debug)]
+pub struct GuardParams {
+    /// `F`: a leg further than this factor from the reference needs credit to print.
+    pub max_deviation_factor: f64,
+    /// `q`: credit needed to qualify, as a fraction of the largest pool's credit.
+    pub min_share_fraction: f64,
+}
+
+impl Default for GuardParams {
+    /// Calibrated against a production replay: `F = 5` sits well above the worst real
+    /// non-qualifying print and well below the weakest observed attack print.
+    ///
+    /// Single source of truth for every call site until these are threaded through
+    /// `configure_me`; changing them requires an `OHLCV_VERSION` bump.
+    fn default() -> Self {
+        Self {
+            max_deviation_factor: 5.0,
+            min_share_fraction: 0.05,
+        }
+    }
+}
+
+/// Per-pool state: reserves and qualification credit.
 #[derive(Clone, Debug)]
 struct PoolState {
     sequence: i64,
     sats: u64,
     token_amount: u64,
-    credit: u64,  // min-depth at last accepted print, seeded at creation
+    /// Min-depth at this pool's last accepted print.
+    credit: u64,
 }
 
-/// The policy engine: stateful fold over legs.
+/// The guard: judges each leg against a reference price derived from pool reserves.
 ///
-/// Maintains:
+/// Reserves are advanced by [`Policy::apply`] as legs stream past, but the starting
-/// - Pool reserves (sats, token_amount, sequence)
+/// state comes from a snapshot (see [`Policy::seeded`]) rather than from whatever legs
-/// - Qualification credit (min-depth at last accepted print)
+/// happen to fall inside the caller's window.  That is what keeps a bucket's verdict
-/// - Reference R (min-depth-weighted median of pool spots)
+/// identical whether it was reached through the raw path, through a rebuild batch, or
+/// through a request for a different timeframe.
 pub struct Policy {
     pools: HashMap<[u8; 32], PoolState>,
-    reference: Option<f64>,  // min-depth-weighted median of pool spots
+    reference: Option<f64>,
     params: GuardParams,
-}
+    /// Reused across `recompute_reference` calls so the hot path does not allocate.
-
+    scratch: Vec<(f64, u64, u64)>,
-#[derive(Clone, Debug)]
-pub struct GuardParams {
-    pub max_deviation_factor: f64,  // F: mute if dev > F AND not exempted
-    pub min_share_fraction: f64,    // q: exemption if summed_credit >= q * largest_credit
 }
 
 impl Policy {
+    /// A policy with no prior knowledge.  Only correct when the caller genuinely starts
+    /// at the token's first trade; otherwise use [`Policy::seeded`].
     pub fn new(params: GuardParams) -> Self {
-        Self {
+        Self::seeded(params, &[])
-            pools: HashMap::new(),
+    }
+
+    /// 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.
+    pub fn seeded(params: GuardParams, snapshot: &[PoolReserves]) -> Self {
+        let mut policy = Self {
+            pools: HashMap::with_capacity(snapshot.len()),
             reference: None,
             params,
+            scratch: Vec::with_capacity(snapshot.len()),
+        };
+
+        for r in snapshot {
+            policy.pools.insert(
+                r.pool,
+                PoolState {
+                    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);
+                }
             }
         }
 
-    /// Judge a leg for acceptance. Must be called before apply().
+        policy
-    pub fn judge(&mut self, leg: &Leg) -> JudgeResult {
+    }
-        // Leg is unpriceable if no token movement.
+
-        if leg.token_delta == 0 {
+    /// Judge a leg.  Pure: call [`Policy::apply`] afterwards to advance state.
+    pub fn judge(&self, leg: &Leg) -> JudgeResult {
+        // A leg that moved no tokens, or bought them for nothing, has no price to print.
+        if leg.token_delta == 0 || leg.sats_delta == 0 {
             return JudgeResult {
                 price: None,
-                accepted: false,
+                verdict: Verdict::Unpriceable,
-                reason: "no token movement".into(),
             };
         }
 
-        let price = (leg.sats_delta.unsigned_abs() as f64) / (leg.token_delta.abs() as f64);
+        let price =
+            leg.sats_delta.unsigned_abs() as f64 / leg.token_delta.unsigned_abs() as f64;
 
-        // If no reference yet, accept (new token, first pool).
+        let Some(reference) = self.reference else {
-        let Some(ref_price) = self.reference else {
             return JudgeResult {
                 price: Some(price),
-                accepted: true,
+                verdict: Verdict::NoReference,
-                reason: "no reference yet".into(),
             };
         };
 
-        // Check deviation from reference.
+        let dev = deviation(price, reference);
-        let dev = (price / ref_price).max(ref_price / price);
-
-        // Tier 1: deviation within F always prints.
         if dev <= self.params.max_deviation_factor {
             return JudgeResult {
                 price: Some(price),
-                accepted: true,
+                verdict: Verdict::WithinBand { dev },
-                reason: format!("dev {:.2} <= F {:.1}", dev, self.params.max_deviation_factor),
             };
         }
 
-        // Tier 2: exemption if this leg's pool credit is high enough.
+        let credit = self.pools.get(&leg.pool).map_or(0, |s| s.credit);
-        let pool_credit = self.pools.get(&leg.pool).map(|ps| ps.credit).unwrap_or(0);
+        let largest_credit = self.pools.values().map(|s| s.credit).max().unwrap_or(0);
-        let largest_credit = self.pools.values().map(|ps| ps.credit).max().unwrap_or(0);
+        // `ceil().max(1)` keeps a pool that has never had an accepted print (credit 0)
-        let threshold_credit = ((largest_credit as f64) * self.params.min_share_fraction) as u64;
+        // 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);
 
-        if largest_credit > 0 && pool_credit >= threshold_credit {
+        let verdict = if credit >= threshold {
-            return JudgeResult {
+            Verdict::CreditExempt {
-                price: Some(price),
-                accepted: true,
-                reason: format!(
-                    "dev {:.2} > F but credit {}/{} >= {:.0}%",
                 dev,
-                    pool_credit,
+                credit,
                 largest_credit,
-                    self.params.min_share_fraction * 100.0
-                ),
-            };
             }
+        } else {
+            Verdict::Muted {
+                dev,
+                credit,
+                largest_credit,
+            }
+        };
 
-        // Muted: deviation too high and credit too low.
         JudgeResult {
             price: Some(price),
-            accepted: false,
+            verdict,
-            reason: format!(
-                "dev {:.2} > F {:.1} AND credit {}/{} < {:.0}%",
-                dev,
-                self.params.max_deviation_factor,
-                pool_credit,
-                largest_credit,
-                self.params.min_share_fraction * 100.0
-            ),
         }
     }
 
-    /// Apply a leg: update pool reserves, credit, and reference.
+    /// Advance state past a leg.  Must be called for every leg, accepted or not: a muted
-    /// Must be called after judge(), regardless of acceptance.
+    /// leg still moved real reserves.
     pub fn apply(&mut self, leg: &Leg, accepted: bool) {
-        // Compute min-depth before borrowing pools (to avoid borrow checker issues).
+        let reference = self.reference;
-        let new_credit = if accepted && leg.token_delta != 0 {
-            let ref_price = self.reference.unwrap_or(1.0);
-            let token_valued = ((leg.token_amount as f64) * ref_price).ceil() as u64;
-            leg.sats.min(token_valued)
-        } else {
-            0
-        };
 
-        // Ensure pool state exists.
+        // `i64::MIN` so a pool's very first leg clears the monotonicity guard below.
-        let state = self
+        let state = self.pools.entry(leg.pool).or_insert(PoolState {
-            .pools
+            sequence: i64::MIN,
-            .entry(leg.pool)
+            sats: 0,
-            .or_insert(PoolState {
+            token_amount: 0,
-                sequence: leg.sequence,
-                sats: leg.sats,
-                token_amount: leg.token_amount,
             credit: 0,
         });
 
-        // Update reserves only if sequence is monotonic (later than what we've seen).
+        if leg.sequence >= state.sequence {
-        if leg.sequence > state.sequence {
             state.sequence = leg.sequence;
             state.sats = leg.sats;
             state.token_amount = leg.token_amount;
 
-            // Update credit on accepted prints.
+            if accepted && leg.token_delta != 0 {
-            if new_credit > 0 {
+                // Value the token side at the reference the leg was judged against; for a
-                state.credit = new_credit;
+                // 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);
             }
         }
 
-        // Recompute reference from all pools' reserves.
+        self.recompute_reference();
-        self.reference = self.compute_reference();
     }
 
-    /// Compute min-depth for a pool: min(sats, tokens * reference).
+    /// Current reference price.
-    /// If no reference yet, use sats (will get updated once reference exists).
-    fn min_depth(&self, sats: u64, token_amount: u64) -> u64 {
-        let ref_price = self.reference.unwrap_or(1.0);
-        let token_valued_at_ref = ((token_amount as f64) * ref_price).ceil() as u64;
-        sats.min(token_valued_at_ref)
-    }
-
-    /// Compute the min-depth-weighted median of pool spot ratios.
-    fn compute_reference(&self) -> Option<f64> {
-        if self.pools.is_empty() {
-            return None;
-        }
-
-        // Filter pools with token_amount >= 10 (avoid divide-by-zero and dust).
-        let mut entries: Vec<(f64, u64)> = self
-            .pools
-            .values()
-            .filter(|ps| ps.token_amount >= 10)
-            .map(|ps| {
-                let spot_ratio = (ps.sats as f64) / (ps.token_amount as f64);
-                let weight = self.min_depth(ps.sats, ps.token_amount);
-                (spot_ratio, weight)
-            })
-            .collect();
-
-        if entries.is_empty() {
-            return None;
-        }
-
-        // Sort by spot ratio ascending.
-        entries.sort_by(|a, b| a.0.partial_cmp(&b.0).unwrap());
-
-        // Compute total weight.
-        let total_weight: u128 = entries.iter().map(|(_, w)| *w as u128).sum();
-
-        // Find the ratio at the 50th percentile.
-        let half_weight = total_weight / 2;
-        let mut cumulative: u128 = 0;
-        for (spot_ratio, weight) in &entries {
-            cumulative += *weight as u128;
-            if cumulative >= half_weight {
-                return Some(*spot_ratio);
-            }
-        }
-
-        // Fallback: last ratio (shouldn't reach here if weights are positive).
-        entries.last().map(|(spot_ratio, _)| *spot_ratio)
-    }
-
-    /// Get current reference price (for testing/debugging).
     pub fn reference(&self) -> Option<f64> {
         self.reference
     }
+
+    /// 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)
+    }
+
+    fn recompute_reference(&mut self) {
+        // Detach the scratch buffer so `self.pools` can be borrowed while filling it.
+        let mut scratch = std::mem::take(&mut self.scratch);
+        scratch.clear();
+        scratch.extend(self.pools.values().filter_map(|s| {
+            spot_ratio(s.sats, s.token_amount).map(|spot| (spot, s.sats, s.token_amount))
+        }));
+        self.reference = weighted_median_reference(&mut scratch);
+        self.scratch = scratch;
+    }
+}
+
+/// 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 {
+        return None;
+    }
+    Some(sats as f64 / token_amount as 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() {
+        return f64::INFINITY;
+    }
+    (price / reference).max(reference / price)
+}
+
+/// `d = min(S, T * R)`: the side of the pool an attacker would have to actually fund.
+///
+/// Valuing the token side at the reference is what stops a pool stuffed with worthless
+/// tokens from out-voting a pool holding real sats.
+fn min_depth(sats: u64, token_amount: u64, reference: f64) -> u64 {
+    let valued = (token_amount as f64 * reference).ceil();
+    if !valued.is_finite() {
+        return sats;
+    }
+    // Rust saturates on out-of-range float-to-int casts, so a huge reference clamps to
+    // u64::MAX and `min` still picks the sats side.
+    sats.min(valued as u64)
+}
+
+/// Min-depth-weighted median of pool spot ratios.
+///
+/// `d_i = min(S_i, T_i * R)` needs an `R` to value the token side, and `R` is what we are
+/// solving for.  Carrying the previous leg's `R` forward would resolve that circularity,
+/// but it would also make the answer depend on where the caller started reading.  Instead
+/// seed with the unweighted median and reweight to a fixed point, which depends on
+/// nothing but the reserves themselves.
+fn weighted_median_reference(entries: &mut [(f64, u64, u64)]) -> Option<f64> {
+    if entries.is_empty() {
+        return None;
+    }
+    entries.sort_unstable_by(|a, b| a.0.partial_cmp(&b.0).unwrap_or(Ordering::Equal));
+
+    let mut reference = median_by(entries, |_| 1)?;
+    for _ in 0..REWEIGHT_PASSES {
+        let Some(next) = median_by(entries, |e| min_depth(e.1, e.2, reference)) else {
+            break;
+        };
+        if next == reference {
+            break;
+        }
+        reference = next;
+    }
+    Some(reference)
+}
+
+/// Lower weighted median of pre-sorted `entries`: the first value whose cumulative weight
+/// reaches half the total.
+///
+/// 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<f64> {
+    let total: u128 = sorted.iter().map(|e| weight(e) as u128).sum();
+    if total == 0 {
+        return sorted.get(sorted.len() / 2).map(|e| e.0);
+    }
+
+    let mut cumulative: u128 = 0;
+    for entry in sorted {
+        cumulative += weight(entry) as u128;
+        if cumulative * 2 >= total {
+            return Some(entry.0);
+        }
+    }
+    sorted.last().map(|e| e.0)
 }
 
 #[cfg(test)]
-mod tests {
+mod tests;
-    use super::*;
-
-    fn leg(
-        pool: u8,
-        ts: i64,
-        seq: i64,
-        sats_delta: i64,
-        token_delta: i64,
-        sats: u64,
-        token_amount: u64,
-    ) -> Leg {
-        Leg {
-            txid: [0u8; 32],
-            pool: [pool; 32],
-            ts,
-            sequence: seq,
-            sats_delta,
-            token_delta,
-            sats,
-            token_amount,
-        }
-    }
-
-    #[test]
-    fn test_unpriceable_leg() {
-        let mut policy = Policy::new(GuardParams {
-            max_deviation_factor: 5.0,
-            min_share_fraction: 0.05,
-        });
-
-        let l = leg(1, 1000, 1, 100, 0, 1000, 1000);
-        let judge = policy.judge(&l);
-        assert!(!judge.accepted);
-        assert!(judge.price.is_none());
-    }
-
-    #[test]
-    fn test_first_leg_always_accepts() {
-        let mut policy = Policy::new(GuardParams {
-            max_deviation_factor: 5.0,
-            min_share_fraction: 0.05,
-        });
-
-        let l = leg(1, 1000, 1, 100, 1000, 1100, 2000);
-        let judge = policy.judge(&l);
-        assert!(judge.accepted, "{}", judge.reason);
-        assert!(judge.price.is_some());
-
-        policy.apply(&l, judge.accepted);
-        assert!(policy.reference().is_some());
-    }
-
-    #[test]
-    fn test_second_leg_within_deviation() {
-        let mut policy = Policy::new(GuardParams {
-            max_deviation_factor: 5.0,
-            min_share_fraction: 0.05,
-        });
-
-        let l1 = leg(1, 1000, 1, 1000, 1000, 2000, 2000);
-        let j1 = policy.judge(&l1);
-        policy.apply(&l1, j1.accepted);
-
-        // Second leg at similar price (ref should be 1.0).
-        let l2 = leg(2, 1000, 2, 950, 1000, 2000, 2000);
-        let j2 = policy.judge(&l2);
-        assert!(j2.accepted, "{}", j2.reason);
-
-        policy.apply(&l2, j2.accepted);
-    }
-
-    #[test]
-    fn test_second_leg_far_from_reference_muted() {
-        let mut policy = Policy::new(GuardParams {
-            max_deviation_factor: 5.0,
-            min_share_fraction: 0.05,
-        });
-
-        let l1 = leg(1, 1000, 1, 1000, 1000, 2000, 2000);
-        let j1 = policy.judge(&l1);
-        policy.apply(&l1, j1.accepted);
-
-        // Second leg at 100x price deviation, no credit (new pool).
-        let l2 = leg(2, 1000, 1, 100000, 1000, 101000, 1001);
-        let j2 = policy.judge(&l2);
-        assert!(!j2.accepted, "{}", j2.reason);
-        assert!(j2.price.is_some());
-
-        policy.apply(&l2, j2.accepted);
-    }
-
-    #[test]
-    fn test_credit_seeding_on_accepted_print() {
-        let mut policy = Policy::new(GuardParams {
-            max_deviation_factor: 5.0,
-            min_share_fraction: 0.05,
-        });
-
-        // Pool 1: large, accepts at ref
-        let l1 = leg(1, 1000, 1, 10000, 1000, 11000, 2000);
-        let j1 = policy.judge(&l1);
-        policy.apply(&l1, j1.accepted);
-
-        // Pool 2: smaller, deviates but has no credit yet (new pool)
-        let l2 = leg(2, 1000, 1, 10000, 1000, 11000, 1000);
-        let j2 = policy.judge(&l2);
-        policy.apply(&l2, j2.accepted);
-
-        // Pool 2 should now have credit from its accepted print.
-        // A later muted print won't restore credit.
-        let l2_muted = leg(2, 1001, 2, 100000, 1000, 101000, 1001);
-        let j2_muted = policy.judge(&l2_muted);
-        assert!(!j2_muted.accepted);
-        policy.apply(&l2_muted, j2_muted.accepted);
-
-        // Verify pool 2's credit didn't rise.
-        let pool2_credit = policy.pools.get(&[2u8; 32]).map(|ps| ps.credit);
-        assert!(pool2_credit.is_some());
-    }
-}

src/db/cauldron/candlestick/policy/tests.rs

@@ -0,0 +1,295 @@
+// Copyright (C) 2024-2026 Whiterun LLC
+//
+// This software is licensed under the GNU Affero General Public License (AGPL), version 3.0 or later.
+// A copy of the license can be found in the LICENSE file or at https://www.gnu.org/licenses/agpl-3.0.html
+
+use super::*;
+
+fn params() -> GuardParams {
+    GuardParams {
+        max_deviation_factor: 5.0,
+        min_share_fraction: 0.05,
+    }
+}
+
+/// A leg at `pool`, leaving that pool holding `sats`/`tokens` afterwards.
+fn leg(
+    pool: u8,
+    sequence: i64,
+    sats_delta: i64,
+    token_delta: i64,
+    sats: u64,
+    token_amount: u64,
+) -> Leg {
+    Leg {
+        txid: [pool; 32],
+        pool: [pool; 32],
+        ts: 1000 + sequence,
+        sequence,
+        sats_delta,
+        token_delta,
+        sats,
+        token_amount,
+    }
+}
+
+fn reserves(pool: u8, sequence: i64, sats: u64, token_amount: u64) -> PoolReserves {
+    PoolReserves {
+        pool: [pool; 32],
+        sequence,
+        sats,
+        token_amount,
+    }
+}
+
+/// 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());
+    judged
+}
+
+#[test]
+fn test_leg_with_no_token_movement_is_unpriceable() {
+    let policy = Policy::new(params());
+    let judged = policy.judge(&leg(1, 1, 100, 0, 1_000_000, 10_000));
+    assert_eq!(judged.verdict, Verdict::Unpriceable);
+    assert!(judged.price.is_none());
+    assert!(!judged.accepted());
+}
+
+/// A leg that moved tokens for zero sats has no meaningful price; printing it as 0.0
+/// would drag a candle's low to the floor.
+#[test]
+fn test_leg_with_no_sats_movement_is_unpriceable() {
+    let policy = Policy::new(params());
+    let judged = policy.judge(&leg(1, 1, 0, -1_000, 1_000_000, 10_000));
+    assert_eq!(judged.verdict, Verdict::Unpriceable);
+    assert!(judged.price.is_none());
+}
+
+#[test]
+fn test_first_leg_of_a_new_token_prints_and_sets_the_reference() {
+    let mut policy = Policy::new(params());
+    let judged = step(&mut policy, &leg(1, 1, 100_000, -1_000, 1_000_000, 10_000));
+
+    assert_eq!(judged.verdict, Verdict::NoReference);
+    assert!(judged.accepted());
+    assert_eq!(judged.price, Some(100.0));
+    assert_eq!(policy.reference(), Some(100.0));
+}
+
+#[test]
+fn test_leg_near_the_reference_prints() {
+    let mut policy = Policy::new(params());
+    step(&mut policy, &leg(1, 1, 100_000, -1_000, 1_000_000, 10_000));
+
+    let judged = step(&mut policy, &leg(2, 2, 190_000, -2_000, 2_000_000, 20_000));
+    assert!(matches!(judged.verdict, Verdict::WithinBand { .. }));
+    assert!(judged.accepted());
+}
+
+#[test]
+fn test_leg_far_from_the_reference_without_credit_is_muted() {
+    let mut policy = Policy::new(params());
+    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);
+    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.
+#[test]
+fn test_pool_earns_credit_on_its_first_accepted_print() {
+    let mut policy = Policy::new(params());
+    step(&mut policy, &leg(1, 1, 100_000, -1_000, 1_000_000, 10_000));
+
+    let second = leg(2, 2, 200_000, -2_000, 2_000_000, 20_000);
+    let judged = step(&mut policy, &second);
+    assert!(judged.accepted(), "{:?}", judged.verdict);
+
+    assert!(
+        policy.credit_of(&[2; 32]) > 0,
+        "a pool must hold credit after its first accepted print, not its second"
+    );
+}
+
+#[test]
+fn test_muted_leg_does_not_earn_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));
+
+    assert_eq!(
+        policy.credit_of(&[9; 32]),
+        0,
+        "a muted print must not qualify the pool that made it"
+    );
+}
+
+/// 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]
+fn test_reference_is_a_function_of_reserves_not_of_history_read() {
+    let mut folded = Policy::new(params());
+    step(&mut folded, &leg(1, 1, 100_000, -1_000, 1_000_000, 10_000));
+    step(&mut folded, &leg(2, 2, 190_000, -2_000, 2_000_000, 20_000));
+    step(&mut folded, &leg(1, 3, 105_000, -1_000, 1_105_000, 9_000));
+
+    // 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)],
+    );
+
+    assert_eq!(
+        folded.reference(),
+        seeded.reference(),
+        "a seeded policy must agree with a folded one on identical reserves"
+    );
+}
+
+/// Corollary: two callers whose windows start at different points must reach the same
+/// verdict on a leg they both see.
+#[test]
+fn test_seeded_and_folded_policies_agree_on_a_shared_leg() {
+    let mut folded = Policy::new(params());
+    step(&mut folded, &leg(1, 1, 100_000, -1_000, 1_000_000, 10_000));
+    step(&mut folded, &leg(2, 2, 190_000, -2_000, 2_000_000, 20_000));
+
+    let seeded = Policy::seeded(
+        params(),
+        &[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);
+    assert_eq!(
+        folded.judge(&shared).accepted(),
+        seeded.judge(&shared).accepted(),
+        "verdict must not depend on where the caller started reading"
+    );
+}
+
+/// Seeding hands credit to pools already trading at market, but withholds it from one
+/// that has been walked far away — otherwise a snapshot would launder an attacker's pool
+/// straight into qualification.
+#[test]
+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
+        ],
+    );
+
+    assert_eq!(policy.reference(), Some(100.0));
+    assert!(policy.credit_of(&[1; 32]) > 0);
+    assert!(policy.credit_of(&[2; 32]) > 0);
+    assert_eq!(
+        policy.credit_of(&[9; 32]),
+        0,
+        "a pool sitting 100x off market must not be seeded with credit"
+    );
+}
+
+/// Min-depth weighting is what stops a pool holding worthless tokens from voting the
+/// reference away from where the real liquidity sits.
+#[test]
+fn test_dust_pool_does_not_move_the_reference() {
+    let policy = Policy::seeded(
+        params(),
+        &[
+            reserves(1, 1, 100_000_000, 1_000_000), // spot 100, deep
+            reserves(2, 2, 200_000_000, 2_000_000), // spot 100, deep
+            reserves(9, 3, 1_000, 10),              // spot 100, but ~nothing behind it
+        ],
+    );
+
+    assert_eq!(policy.reference(), Some(100.0));
+}
+
+#[test]
+fn test_pools_below_the_dust_floor_are_ignored_entirely() {
+    let policy = Policy::seeded(
+        params(),
+        &[
+            reserves(1, 1, 1_000_000, 10_000), // spot 100
+            reserves(9, 2, 1_000_000, 1),      // spot 1_000_000, below MIN_TOKEN_RESERVE
+        ],
+    );
+    assert_eq!(policy.reference(), Some(100.0));
+}
+
+#[test]
+fn test_policy_with_no_usable_pools_has_no_reference() {
+    let policy = Policy::seeded(params(), &[reserves(1, 1, 0, 0)]);
+    assert_eq!(policy.reference(), None);
+}
+
+/// `cumulative >= total / 2` truncates, which tips two-pool sets towards the lower ratio.
+/// The comparison is written as `2 * cumulative >= total` to avoid that.
+#[test]
+fn test_weighted_median_is_not_biased_by_integer_division() {
+    let mut entries = vec![(100.0, 3, 1), (200.0, 3, 1)];
+    let picked = weighted_median_reference(&mut entries).unwrap();
+    assert!(
+        picked == 100.0 || picked == 200.0,
+        "expected one of the two inputs, got {picked}"
+    );
+
+    // An odd, clearly-weighted set must land on the true weighted median.
+    let mut skewed = vec![(10.0, 1, 1), (100.0, 1_000_000, 1_000_000), (1000.0, 1, 1)];
+    assert_eq!(weighted_median_reference(&mut skewed), Some(100.0));
+}
+
+/// All-zero weights used to make the cumulative test pass on the first entry, returning
+/// the smallest ratio rather than the middle one.
+#[test]
+fn test_all_zero_weights_fall_back_to_the_positional_median() {
+    let mut entries = vec![(10.0, 0, 0), (100.0, 0, 0), (1000.0, 0, 0)];
+    assert_eq!(weighted_median_reference(&mut entries), Some(100.0));
+}
+
+#[test]
+fn test_min_depth_values_the_token_side_at_the_reference() {
+    // Token side is the thinner one: 10 tokens * 100 = 1_000 < 50_000 sats.
+    assert_eq!(min_depth(50_000, 10, 100.0), 1_000);
+    // Sats side is the thinner one.
+    assert_eq!(min_depth(500, 10, 100.0), 500);
+}
+
+#[test]
+fn test_min_depth_saturates_instead_of_overflowing() {
+    assert_eq!(min_depth(42, u64::MAX, f64::MAX), 42);
+}
+
+#[test]
+fn test_deviation_is_symmetric_and_rejects_degenerate_input() {
+    assert_eq!(deviation(200.0, 100.0), 2.0);
+    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);
+}
+
+/// A leg replayed at or below a pool's known sequence must not rewind its reserves.
+#[test]
+fn test_out_of_order_leg_does_not_rewind_reserves() {
+    let mut policy = Policy::new(params());
+    step(&mut policy, &leg(1, 10, 100_000, -1_000, 1_000_000, 10_000));
+    let after_current = policy.reference();
+
+    step(&mut policy, &leg(1, 5, 100_000, -1_000, 7, 7));
+    assert_eq!(
+        policy.reference(),
+        after_current,
+        "a stale leg must not overwrite newer reserves"
+    );
+}

src/db/cauldron/candlestick/tests.rs

@@ -144,8 +144,11 @@ async fn insert_multileg_trade_at(
 /// Dividing the signed sums produced a price no leg traded at — on mainnet token NWB
 /// (tx 1E84F4E9…1916, 27 legs) that printed 30,792,599.5 sats/unit against legs that
 /// actually executed between 0.288 and 0.335.
+///
+/// The durable invariant is that a printed price must be one a leg actually executed at.
+/// Under per-leg pricing the close is specifically the last accepted leg's price.
 #[tokio::test]
-async fn test_multipool_arb_priced_by_gross_volume_not_net() {
+async fn test_multipool_arb_prices_from_its_legs_not_their_net() {
     let db = mock_db_pool(setup_db).await;
     let token = TokenID::from_byte_array([0xC1; 32]);
     let token_blob = token.to_blob();
@@ -178,14 +181,16 @@ async fn test_multipool_arb_priced_by_gross_volume_not_net() {
     );
 }
 
-/// Every leg pointing the same way is the ordinary case: gross and net agree exactly,
+/// Every leg pointing the same way is the ordinary case — the one covering 99.76% of
-/// so 99.76% of mainnet prints — including the OLA supply-shock crash — are untouched.
+/// mainnet prints, including the OLA supply-shock crash. Nothing here should be muted or
+/// distorted: each leg executed at a genuine price and the candle closes on the last.
 #[tokio::test]
-async fn test_single_direction_multileg_price_matches_net_ratio() {
+async fn test_single_direction_multileg_closes_on_its_last_leg() {
     let db = mock_db_pool(setup_db).await;
     let token = TokenID::from_byte_array([0xC2; 32]);
     let token_blob = token.to_blob();
 
+    // Leg prices: 50.0, 50.0, 49.5 — a tight, ordinary spread.
     let legs = [(150_000i64, -3_000i64), (50_000, -1_000), (99_000, -2_000)];
     let mut conn = db.cauldron_w.acquire().await.unwrap();
     insert_multileg_trade_at(&mut conn, &token, 0x22, 1000, &legs).await;
@@ -195,15 +200,46 @@ async fn test_single_direction_multileg_price_matches_net_ratio() {
         .unwrap()
         .expect("router transaction must price");
 
-    // Per-leg pricing: the close is the last leg's price, not the per-tx net ratio.
+    let leg_prices: Vec<f64> = legs
-    // Last leg: (99_000, -2_000) = 99,000 / 2,000 = 49.5
+        .iter()
-    let last_leg_price = 99_000.0 / 2_000.0;
+        .map(|(s, t)| *s as f64 / t.unsigned_abs() as f64)
+        .collect();
+    let lowest = leg_prices.iter().cloned().fold(f64::MAX, f64::min);
+    let highest = leg_prices.iter().cloned().fold(f64::MIN, f64::max);
+    assert!(
+        price >= lowest && price <= highest,
+        "price {price} must be one the legs actually executed at, within [{lowest}, {highest}]"
+    );
+
+    let last_leg_price = *leg_prices.last().unwrap();
     assert!(
         (price - last_leg_price).abs() < f64::EPSILON,
-        "close should be last leg's price: {price} vs {last_leg_price}"
+        "close should be the last leg's price: {price} vs {last_leg_price}"
     );
 }
 
+/// Regression: the seed lookback was briefly capped at 24 hours, which silently dropped
+/// the carry-forward price for any token trading less often than daily and left a hole
+/// where the leading candles should be.
+#[tokio::test]
+async fn test_seed_lookback_reaches_past_a_long_silence() {
+    let db = mock_db_pool(setup_db).await;
+    let token = TokenID::from_byte_array([0xC5; 32]);
+    let token_blob = token.to_blob();
+
+    let mut conn = db.cauldron_w.acquire().await.unwrap();
+    insert_trade_at(&mut conn, &token, 0x31, 1000, 100_000, 2_000).await;
+
+    // Ask 60 days later. The only trade is far outside any 24-hour window.
+    let sixty_days = 1000 + 60 * 24 * 3600;
+    let price = super::fetch_last_close_before(&db.cauldron_r, &token_blob, sixty_days)
+        .await
+        .unwrap()
+        .expect("a dormant token must still carry its last known price");
+
+    assert!((price - 50.0).abs() < f64::EPSILON, "got {price}");
+}
+
 /// A transaction whose legs cancel exactly used to print nothing at all (the chart
 /// carried the previous close). Its legs are real executions and now price normally.
 #[tokio::test]

src/db/cauldron/ohlcv.rs

@@ -13,7 +13,9 @@ use sqlx::{Row, SqlitePool};
 ///
 /// 2: price switched from the signed net ratio to the gross volume ratio.
 /// 3: pricing switched from per-transaction to per-leg with policy-based acceptance.
-pub const OHLCV_VERSION: u32 = 3;
+/// 4: policy reference seeded from reserves rather than folded from the batch start, and
+///    buckets with no priceable leg are skipped instead of written with f64 sentinels.
+pub const OHLCV_VERSION: u32 = 4;
 const OHLCV_VERSION_KEY: &str = "ohlcv_version";
 
 pub async fn create_table(pool: &SqlitePool) {
@@ -98,7 +100,9 @@ pub async fn rebuild_range(
     since_ts: i64,
     until_ts: i64,
 ) -> Result<u64> {
-    use crate::db::cauldron::candlestick::{GuardParams, Leg, Policy};
+    use crate::db::cauldron::candlestick::{
+        fetch_reserve_snapshot, to_hash32, GuardParams, Leg, Policy,
+    };
     use std::collections::HashMap;
 
     if since_ts >= until_ts {
@@ -136,15 +140,12 @@ ORDER BY phe.token_id ASC, phe.effective_timestamp ASC, phe.sequence ASC;
     }
 
     // Phase 2: fold through legs per-token using the policy core to compute buckets.
-    let params = GuardParams {
+    let params = GuardParams::default();
-        max_deviation_factor: 5.0,
-        min_share_fraction: 0.05,
-    };
 
     let mut buckets: HashMap<(Vec<u8>, i64), OhlcvBucket> = HashMap::new();
 
     let mut current_token: Option<Vec<u8>> = None;
-    let mut policy = Policy::new(params.clone());
+    let mut policy = Policy::new(params);
 
     for row in &rows {
         let token_id: Vec<u8> = row.get(0);
@@ -153,21 +154,19 @@ ORDER BY phe.token_id ASC, phe.effective_timestamp ASC, phe.sequence ASC;
         let ts: i64 = row.get(8);
         let bucket_ts = (ts / 3600) * 3600;
 
-        // Reset policy when we move to a new token.
+        // A new token starts a new policy — seeded from that token's reserves as they
+        // stood when this batch opened.  Callers materialise history in 24-hour batches,
+        // so an unseeded policy would forget everything at every batch boundary and wave
+        // through the first leg of every day of history unjudged.
         if current_token.as_ref() != Some(&token_id) {
             current_token = Some(token_id.clone());
-            policy = Policy::new(params.clone());
+            let snapshot = fetch_reserve_snapshot(read_pool, &token_id, since_ts).await?;
+            policy = Policy::seeded(params, &snapshot);
         }
 
-        // Build the Leg struct.
-        let mut txid_arr = [0u8; 32];
-        let mut pool_arr = [0u8; 32];
-        txid_arr.copy_from_slice(&txid);
-        pool_arr.copy_from_slice(&pool_bytes);
-
         let leg = Leg {
-            txid: txid_arr,
+            txid: to_hash32(&txid)?,
-            pool: pool_arr,
+            pool: to_hash32(&pool_bytes)?,
             ts,
             sequence: row.get(7),
             sats_delta: row.get(3),
@@ -180,7 +179,7 @@ ORDER BY phe.token_id ASC, phe.effective_timestamp ASC, phe.sequence ASC;
         let judge = policy.judge(&leg);
 
         let sats_delta_abs = leg.sats_delta.unsigned_abs() as i64;
-        let token_delta_abs = leg.token_delta.abs() as i64;
+        let token_delta_abs = leg.token_delta.unsigned_abs() as i64;
 
         // Get or create the bucket for this (token, bucket_ts).
         let bucket_key = (token_id.clone(), bucket_ts);
@@ -189,10 +188,10 @@ ORDER BY phe.token_id ASC, phe.effective_timestamp ASC, phe.sequence ASC;
         // Always accumulate volume.
         bucket.volume_sats += sats_delta_abs;
         bucket.volume_tokens += token_delta_abs;
-        bucket.txids.insert(txid_arr);
+        bucket.txids.insert(leg.txid);
 
         // Update OHLC only if accepted and priceable.
-        if judge.accepted {
+        if judge.accepted() {
             if let Some(price) = judge.price {
                 if bucket.first_accepted_price.is_none() {
                     bucket.first_accepted_price = Some(price);
@@ -203,7 +202,7 @@ ORDER BY phe.token_id ASC, phe.effective_timestamp ASC, phe.sequence ASC;
             }
         }
 
-        policy.apply(&leg, judge.accepted);
+        policy.apply(&leg, judge.accepted());
     }
 
     // Phase 3: insert all computed buckets in one transaction.
@@ -211,8 +210,15 @@ ORDER BY phe.token_id ASC, phe.effective_timestamp ASC, phe.sequence ASC;
     let mut inserted = 0u64;
 
     for ((token_id, bucket_ts), bucket) in buckets {
-        let open = bucket.first_accepted_price.unwrap_or(bucket.last_accepted_price.unwrap_or(0.0));
+        // A bucket whose legs were all muted or unpriceable has no OHLC to report: its
-        let close = bucket.last_accepted_price.unwrap_or(bucket.first_accepted_price.unwrap_or(0.0));
+        // `high`/`low` are still the f64::MIN/MAX sentinels and open/close are unset.
+        // Writing that row would put ±1.8e308 into the table and hand it straight to the
+        // chart. Skip it, exactly as the SQL this replaced did by inner-joining prices to
+        // volume; the raw path's carry-forward renders the gap from the previous close.
+        let (Some(open), Some(close)) = (bucket.first_accepted_price, bucket.last_accepted_price)
+        else {
+            continue;
+        };
 
         inserted += sqlx::query(
             "INSERT OR IGNORE INTO ohlcv_1h
@@ -505,6 +511,36 @@ mod tests {
         assert_eq!(n, 0, "mempool trades must not be materialised");
     }
 
+    /// Regression: a bucket whose legs are all unpriceable used to be written out with its
+    /// `high`/`low` accumulators still at the `f64::MIN`/`f64::MAX` sentinels, putting
+    /// ±1.8e308 into the table and from there straight onto the chart. Such a bucket has
+    /// no OHLC to report and must simply not be materialised.
+    #[tokio::test]
+    async fn test_rebuild_range_skips_buckets_with_no_priceable_leg() {
+        let pool = test_pool().await;
+        setup_db(&pool).await;
+
+        // A confirmed leg that moves sats but no tokens: real volume, no price.
+        insert_confirmed_trade(
+            &pool, [0x01; 32], [0x02; 32], [0x03; 32], 1727963400, -1000, 0,
+        )
+        .await;
+
+        let n = rebuild_range(&pool, &pool, 1727960400, 1727967600)
+            .await
+            .unwrap();
+        assert_eq!(n, 0, "a bucket with no priceable leg must not be written");
+
+        let rows: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM ohlcv_1h")
+            .fetch_one(&pool)
+            .await
+            .unwrap();
+        assert_eq!(
+            rows, 0,
+            "no row should exist, let alone one carrying f64 sentinels"
+        );
+    }
+
     /// A multi-pool arbitrage transaction whose legs nearly cancel must materialise the
     /// price its legs executed at, not the signed-net ratio.
     #[tokio::test]

src/rpc/candlesticks/tests.rs

@@ -66,6 +66,7 @@ async fn setup_mock_db(pool: sqlx::SqlitePool) {
     utxo_funding::create_table(&pool).await;
     tx::create_table(&pool).await;
     pool::create_table(&pool).await;
+    ohlcv::create_table(&pool).await;
     dummy_init_seq();
 
     let mut conn = pool.acquire().await.unwrap();
@@ -380,6 +381,7 @@ async fn test_single_swap_multiple_pools() {
         utxo_funding::create_table(&pool).await;
         tx::create_table(&pool).await;
         pool::create_table(&pool).await;
+        ohlcv::create_table(&pool).await;
         dummy_init_seq();
 
         let mut conn = pool.acquire().await.unwrap();
@@ -481,6 +483,7 @@ async fn test_candle_carries_price_when_net_zero_tokens_but_has_volume() {
         utxo_funding::create_table(&pool).await;
         tx::create_table(&pool).await;
         pool::create_table(&pool).await;
+        ohlcv::create_table(&pool).await;
         dummy_init_seq();
 
         let mut conn = pool.acquire().await.unwrap();