Stage 2 Phase 2: Integrate per-leg pricing with policy core

Rewrites the core pricing functions to evaluate each leg individually through
the policy engine rather than grouping by transaction and computing net ratios.

Changes:
- fetch_raw_trades → fetch_raw_legs: returns per-leg data with pool info,
  deltas, post-state reserves, and sequence numbers
- aggregate_raw_trades: now takes per-leg data and folds through the policy
  for each leg; judges acceptance, updates state, and accumulates OHLC
- candlesticks: creates a Policy instance with default params (F=5, q=5%),
  passes it through the aggregation pipeline

Key behaviors:
- Volume ALWAYS counted (both accepted and muted legs)
- OHLC updated ONLY for accepted legs
- transaction_count = unique txids in interval
- Carry-forward logic for intervals with no accepted prints
- Policy state maintained and updated per-leg across the full window

Tests: 222 passing (5 new policy tests + 217 existing candle/ohlcv/price tests)

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

src/db/cauldron/candlestick/mod.rs

@@ -10,6 +10,8 @@ use bitcoincash::TokenID;
 use serde::Serialize;
 use sqlx::{Row, SqlitePool};
 
+pub use self::policy::{GuardParams, Leg, Policy};
+
 #[derive(Debug, Serialize)]
 pub struct CandlestickData {
     pub time: i64, // start of the interval
@@ -71,14 +73,16 @@ impl PriceInterval {
 }
 
 fn aggregate_raw_trades(
-    all_trades: &[(i64, i64, i64)],
+    all_legs: &[Leg],
     intervals: Vec<PriceInterval>,
     step_size: i64,
     mut found_first_trade: bool,
     mut last_close_price: Option<f64>,
+    policy: &mut Policy,
 ) -> (Vec<CandlestickData>, bool, Option<f64>) {
     let mut result = Vec::with_capacity(intervals.len());
-    let mut trade_index = 0;
+    let mut leg_index = 0;
+    let mut txid_set = std::collections::HashSet::new();
 
     for interval in intervals {
         let interval_start = interval.start;
@@ -86,38 +90,51 @@ fn aggregate_raw_trades(
         let mut pi = PriceInterval::new(interval_start, step_size);
         let mut first_trade_in_interval = true;
 
-        while trade_index < all_trades.len() {
+        while leg_index < all_legs.len() {
-            let (ts, vol_sats, vol_tokens) = all_trades[trade_index];
+            let leg = &all_legs[leg_index];
-            if ts < interval_start {
+            if leg.ts < interval_start {
-                trade_index += 1;
+                leg_index += 1;
                 continue;
             }
-            if ts >= interval_end {
+            if leg.ts >= interval_end {
                 break;
             }
 
-            if vol_tokens != 0 {
+            let judge = policy.judge(leg);
-                let price = vol_sats as f64 / vol_tokens as f64;
+            let sats_delta_abs = leg.sats_delta.unsigned_abs() as i64;
-                if first_trade_in_interval {
+            let token_delta_abs = leg.token_delta.abs() as i64;
-                    pi.open = Some(price);
+
-                    pi.high = price;
+            // Always count volume, regardless of acceptance.
-                    pi.low = price;
+            pi.volume_sats += sats_delta_abs;
-                    first_trade_in_interval = false;
+            pi.volume_tokens += token_delta_abs;
-                }
+            txid_set.insert(leg.txid);
-                pi.close = Some(price);
+
-                if price.is_finite() {
+            // Update OHLC only if accepted and priceable.
-                    pi.high = pi.high.max(price);
+            if judge.accepted {
-                    pi.low = pi.low.min(price);
+                if let Some(price) = judge.price {
+                    if first_trade_in_interval {
+                        pi.open = Some(price);
+                        pi.high = price;
+                        pi.low = price;
+                        first_trade_in_interval = false;
+                    }
+                    pi.close = Some(price);
+                    if price.is_finite() {
+                        pi.high = pi.high.max(price);
+                        pi.low = pi.low.min(price);
+                    }
                 }
             }
 
-            pi.volume_sats += vol_sats;
+            policy.apply(leg, judge.accepted);
-            pi.volume_tokens += vol_tokens;
+            leg_index += 1;
-            pi.transaction_count += 1;
-            trade_index += 1;
         }
 
-        // Carry forward last close when volume exists but net tokens are zero.
+        // Set tx_count to unique txids in this interval.
+        pi.transaction_count = txid_set.len() as i64;
+        txid_set.clear();
+
+        // Carry forward last close when no accepted prints in this interval.
         if pi.transaction_count > 0 && (pi.open.is_none() || pi.close.is_none()) {
             if let Some(prev) = last_close_price {
                 if pi.open.is_none() {
@@ -206,33 +223,29 @@ fn fill_ohlcv_candles(
     (result, found_first_trade, last_close)
 }
 
-/// Returns one row per transaction: `(effective_timestamp, volume_sats, volume_tokens)`.
+/// Returns per-leg data for policy evaluation.
-///
+/// Each row is: (txid, effective_timestamp, pool, sats_delta, token_delta, sats, token_amount, sequence)
-/// Volumes are gross sums of the absolute per-leg deltas, so the derived price
+async fn fetch_raw_legs(
-/// `volume_sats / volume_tokens` is the volume-weighted average of the prices actually
-/// executed by that transaction's legs, and is therefore always bounded by the cheapest
-/// and dearest leg. Summing the *signed* deltas instead lets a multi-pool arbitrage
-/// transaction — which buys from one pool and sells into others — cancel almost all of
-/// its token movement and divide real satoshis by a near-zero remainder, fabricating a
-/// price no leg ever traded at.
-async fn fetch_raw_trades(
     pool: &SqlitePool,
     token_blob: &[u8],
     timestamp_start: i64,
     timestamp_end: i64,
-) -> Result<Vec<(i64, i64, i64)>> {
+) -> Result<Vec<Leg>> {
     let sql = r#"
 SELECT
+  phe.txid,
   phe.effective_timestamp,
-  SUM(ABS(phe.sats_delta))  AS volume_sats,
+  phe.pool,
-  SUM(ABS(phe.token_delta)) AS volume_tokens,
+  phe.sats_delta,
-  MIN(phe.sequence)         AS min_sequence
+  phe.token_delta,
+  phe.sats,
+  phe.token_amount,
+  phe.sequence
 FROM pool_history_entry AS phe
 WHERE phe.token_id = ?
   AND phe.effective_timestamp >= ?
   AND phe.effective_timestamp < ?
-GROUP BY phe.txid, phe.effective_timestamp
+ORDER BY phe.effective_timestamp ASC, phe.sequence ASC;
-ORDER BY phe.effective_timestamp ASC, min_sequence ASC;
 "#;
     let rows = sqlx::query(sql)
         .bind(token_blob)
@@ -243,7 +256,24 @@ ORDER BY phe.effective_timestamp ASC, min_sequence ASC;
 
     Ok(rows
         .into_iter()
-        .map(|r| (r.get(0), r.get(1), r.get(2)))
+        .map(|r| {
+            let txid: Vec<u8> = r.get(0);
+            let pool_bytes: Vec<u8> = r.get(2);
+            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);
+            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())
 }
 
@@ -291,6 +321,13 @@ pub async fn candlesticks(
 
     let token_blob = display_hex_to_blob::<TokenID>(token_id)?;
 
+    // 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
     // for any overlapping period.
@@ -319,7 +356,7 @@ pub async fn candlesticks(
 
         if ohlcv_end < timestamp_end {
             // Tail: query raw for [ohlcv_end, timestamp_end) and append.
-            let raw_trades = fetch_raw_trades(pool, &token_blob, ohlcv_end, timestamp_end).await?;
+            let raw_legs = fetch_raw_legs(pool, &token_blob, ohlcv_end, timestamp_end).await?;
 
             let mut tail_intervals = Vec::new();
             let mut t = ohlcv_end;
@@ -329,11 +366,12 @@ pub async fn candlesticks(
             }
 
             let (tail, _, _) = aggregate_raw_trades(
-                &raw_trades,
+                &raw_legs,
                 tail_intervals,
                 step_size,
                 found_first,
                 last_close,
+                &mut policy,
             );
             result.extend(tail);
         }
@@ -349,10 +387,10 @@ pub async fn candlesticks(
         current_start += step_size;
     }
 
-    let all_trades = fetch_raw_trades(pool, &token_blob, timestamp_start, timestamp_end).await?;
+    let all_legs = fetch_raw_legs(pool, &token_blob, timestamp_start, timestamp_end).await?;
 
     let (result, _, _) =
-        aggregate_raw_trades(&all_trades, intervals, step_size, seed_found, seed_close);
+        aggregate_raw_trades(&all_legs, intervals, step_size, seed_found, seed_close, &mut policy);
     Ok(result)
 }