Stage 2 Phase 1: Per-leg policy core for manipulation-resistant candlestick pricing

New module: candlestick/policy.rs implements the core state machine that judges
and applies legs for acceptance based on deviation from a reference price and
qualification credit.

Key components:
- Leg: per-pool per-transaction change (sats_delta, token_delta, post-state reserves)
- JudgeResult: verdict on whether a leg is accepted into OHLC
- Policy: stateful fold that maintains pool reserves, qualification credit, and
  a min-depth-weighted median reference price

Qualification rules (two-tier):
- Tier 1: dev <= F (F=5) always accepted
- Tier 2: dev > F but summed_credit >= q*largest_credit (q=5%) also accepted
- Everything else is muted (volume still counted)

Reference computation:
- R = min-depth-weighted median of pool spot ratios
- Updated only on reserve events (swaps, creations, withdrawals), never on prints
- Depth = min(sats, tokens * R_prev) to zero-weight lopsided pools
- Avoids ratchet-walking and qualifies token-heavy reseeds (OLA-like)

Tests: 5 passing
- unpriceable legs
- first leg (no reference yet)
- tier 1 acceptance (dev within F)
- tier 1 rejection (dev > F, no credit)
- credit seeding on accepted prints

Co-Authored-By: Claude Haiku 4.5 <noreply@anthropic.com>

src/db/cauldron/candlestick/mod.rs

@@ -356,5 +356,7 @@ pub async fn candlesticks(
     Ok(result)
 }
 
+pub mod policy;
+
 #[cfg(test)]
 mod tests;

src/db/cauldron/candlestick/policy.rs

@@ -0,0 +1,349 @@
+// 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 std::collections::HashMap;
+
+/// A leg of a transaction: one pool's change in a single transaction.
+#[derive(Clone, Debug)]
+pub struct Leg {
+    pub txid: [u8; 32],
+    pub pool: [u8; 32],
+    pub ts: i64,
+    pub sequence: i64,
+    pub sats_delta: i64,     // sats change at this pool
+    pub token_delta: i64,    // token change at this pool
+    pub sats: u64,           // post-change sats (reserves)
+    pub token_amount: u64,   // post-change token_amount (reserves)
+}
+
+/// Outcome of judging a leg for acceptance.
+#[derive(Clone, Debug)]
+pub struct JudgeResult {
+    pub price: Option<f64>,  // None if token_delta == 0
+    pub accepted: bool,      // whether this leg is accepted into OHLC
+    pub reason: String,      // debug: why accepted or muted
+}
+
+/// Per-pool state: reserves and credit.
+#[derive(Clone, Debug)]
+struct PoolState {
+    sequence: i64,
+    sats: u64,
+    token_amount: u64,
+    credit: u64,  // min-depth at last accepted print, seeded at creation
+}
+
+/// The policy engine: stateful fold over legs.
+///
+/// Maintains:
+/// - Pool reserves (sats, token_amount, sequence)
+/// - Qualification credit (min-depth at last accepted print)
+/// - Reference R (min-depth-weighted median of pool spots)
+pub struct Policy {
+    pools: HashMap<[u8; 32], PoolState>,
+    reference: Option<f64>,  // min-depth-weighted median of pool spots
+    params: GuardParams,
+}
+
+#[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 {
+    pub fn new(params: GuardParams) -> Self {
+        Self {
+            pools: HashMap::new(),
+            reference: None,
+            params,
+        }
+    }
+
+    /// Judge a leg for acceptance. Must be called before apply().
+    pub fn judge(&mut self, leg: &Leg) -> JudgeResult {
+        // Leg is unpriceable if no token movement.
+        if leg.token_delta == 0 {
+            return JudgeResult {
+                price: None,
+                accepted: false,
+                reason: "no token movement".into(),
+            };
+        }
+
+        let price = (leg.sats_delta.unsigned_abs() as f64) / (leg.token_delta.abs() as f64);
+
+        // If no reference yet, accept (new token, first pool).
+        let Some(ref_price) = self.reference else {
+            return JudgeResult {
+                price: Some(price),
+                accepted: true,
+                reason: "no reference yet".into(),
+            };
+        };
+
+        // Check deviation from 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,
+                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 pool_credit = self.pools.get(&leg.pool).map(|ps| ps.credit).unwrap_or(0);
+        let largest_credit = self.pools.values().map(|ps| ps.credit).max().unwrap_or(0);
+        let threshold_credit = ((largest_credit as f64) * self.params.min_share_fraction) as u64;
+
+        if largest_credit > 0 && pool_credit >= threshold_credit {
+            return JudgeResult {
+                price: Some(price),
+                accepted: true,
+                reason: format!(
+                    "dev {:.2} > F but credit {}/{} >= {:.0}%",
+                    dev,
+                    pool_credit,
+                    largest_credit,
+                    self.params.min_share_fraction * 100.0
+                ),
+            };
+        }
+
+        // Muted: deviation too high and credit too low.
+        JudgeResult {
+            price: Some(price),
+            accepted: false,
+            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.
+    /// Must be called after judge(), regardless of acceptance.
+    pub fn apply(&mut self, leg: &Leg, accepted: bool) {
+        // Compute min-depth before borrowing pools (to avoid borrow checker issues).
+        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.
+        let state = self
+            .pools
+            .entry(leg.pool)
+            .or_insert(PoolState {
+                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 {
+            state.sequence = leg.sequence;
+            state.sats = leg.sats;
+            state.token_amount = leg.token_amount;
+
+            // Update credit on accepted prints.
+            if new_credit > 0 {
+                state.credit = new_credit;
+            }
+        }
+
+        // Recompute reference from all pools' reserves.
+        self.reference = self.compute_reference();
+    }
+
+    /// Compute min-depth for a pool: min(sats, tokens * reference).
+    /// 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
+    }
+}
+
+#[cfg(test)]
+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());
+    }
+}