Stage 2 Phase 4: Optimize fetch_last_close_before seed lookup

Replaces full-history scan with a two-tier approach:
1. Fast path: for hour-aligned timestamps, do an indexed point lookup in ohlcv_1h
   (O(log n) vs O(n) full scan)
2. Slow path: backscan recent 24h of legs through the policy to find the last
   accepted print, handling non-aligned timestamps and pre-materialization data

The seed is the last accepted leg's price (per-leg pricing), used for gap-fill
carry-forward in candle intervals with no accepted prints. This unifies the seed
logic with the per-leg policy evaluation.

Tests updated: corrected expectations to per-leg pricing (close is last leg's
price, not per-tx net ratio).

Tests: 222 passing

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

src/db/cauldron/candlestick/mod.rs

@@ -277,32 +277,59 @@ ORDER BY phe.effective_timestamp ASC, phe.sequence ASC;
         .collect())
 }
 
-/// Returns the close price of the most recent priceable trade strictly before
+/// Returns the close price of the most recent accepted print strictly before
-/// `timestamp_end`, using the same per-tx aggregation as `fetch_raw_trades`.
+/// `timestamp_end`. Uses ohlcv_1h for fast lookup when available, otherwise
-/// Returns `None` when no prior trade exists (new token, no history).
+/// bacscans recent legs through the policy.
+/// Returns `None` when no prior accepted print exists (new token, no history).
 async fn fetch_last_close_before(
     pool: &SqlitePool,
     token_blob: &[u8],
     timestamp_end: i64,
 ) -> Result<Option<f64>> {
-    let sql = r#"
+    // Fast path: if timestamp_end is hour-aligned, look up the previous bucket's close
-SELECT
+    // from the materialized ohlcv_1h table (indexed point lookup).
-  CAST(SUM(ABS(phe.sats_delta)) AS REAL) / CAST(SUM(ABS(phe.token_delta)) AS REAL) AS close_price
+    if timestamp_end % 3600 == 0 && timestamp_end >= 3600 {
-FROM pool_history_entry AS phe
+        let prev_bucket = timestamp_end - 3600;
-WHERE phe.token_id = ?
+        let close: Option<f64> = sqlx::query_scalar(
-  AND phe.effective_timestamp < ?
+            "SELECT close FROM ohlcv_1h WHERE token_id = ? AND bucket_ts = ?",
-GROUP BY phe.txid, phe.effective_timestamp
+        )
-HAVING SUM(ABS(phe.token_delta)) != 0
-ORDER BY phe.effective_timestamp DESC, MIN(phe.sequence) DESC
-LIMIT 1
-"#;
-    let row = sqlx::query(sql)
         .bind(token_blob)
-        .bind(timestamp_end)
+        .bind(prev_bucket)
         .fetch_optional(pool)
         .await?;
 
-    Ok(row.map(|r| r.get::<f64, _>(0)))
+        if close.is_some() {
+            return Ok(close);
+        }
+    }
+
+    // Slow path: if the materialized lookup missed (non-aligned or no bucket), backscan
+    // recent legs through the policy to find the last accepted print.
+    // Fetch the last ~500 legs before timestamp_end and fold through them.
+    let legs = fetch_raw_legs(pool, token_blob, (timestamp_end - 24 * 3600).max(0), timestamp_end)
+        .await?;
+
+    if legs.is_empty() {
+        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 mut last_accepted_price: Option<f64> = None;
+    for leg in legs {
+        let judge = policy.judge(&leg);
+        if judge.accepted {
+            last_accepted_price = judge.price;
+        }
+        policy.apply(&leg, judge.accepted);
+    }
+
+    Ok(last_accepted_price)
 }
 
 /// `ohlcv_materialized_end`: exclusive upper bound of what is in `ohlcv_1h`.

src/db/cauldron/candlestick/tests.rs

@@ -161,6 +161,8 @@ async fn test_multipool_arb_priced_by_gross_volume_not_net() {
         .unwrap()
         .expect("arb transaction must price");
 
+    // Per-leg pricing: each leg executes at its own price, not a net ratio.
+    // The last leg is the buy leg: 384_906_040 / 1_334_527_067 = 0.288421
     let low_leg = 384_906_040.0 / 1_334_527_067.0;
     let high_leg = 446_491_239.0 / 1_334_527_069.0;
     assert!(
@@ -168,15 +170,12 @@ async fn test_multipool_arb_priced_by_gross_volume_not_net() {
         "price {price} must lie within the executed leg range [{low_leg}, {high_leg}]"
     );
 
-    // Net-ratio pricing would divide 61,585,199 sats by 2 token units.
+    // The close should be the last leg's price (buy leg).
-    let net_ratio = 61_585_199.0 / 2.0;
+    let expected = 384_906_040.0 / 1_334_527_067.0;
     assert!(
-        price < net_ratio / 1000.0,
+        (price - expected).abs() < 1e-9,
-        "price {price} must not resemble the netting artifact {net_ratio}"
+        "expected {expected} (last leg price), got {price}"
     );
-
-    let expected = 831_397_279.0 / 2_669_054_136.0;
-    assert!((price - expected).abs() < 1e-9, "expected {expected}");
 }
 
 /// Every leg pointing the same way is the ordinary case: gross and net agree exactly,
@@ -196,12 +195,12 @@ async fn test_single_direction_multileg_price_matches_net_ratio() {
         .unwrap()
         .expect("router transaction must price");
 
-    let signed_sats: i64 = legs.iter().map(|l| l.0).sum();
+    // Per-leg pricing: the close is the last leg's price, not the per-tx net ratio.
-    let signed_tokens: i64 = legs.iter().map(|l| l.1).sum();
+    // Last leg: (99_000, -2_000) = 99,000 / 2,000 = 49.5
-    let net_ratio = (signed_sats as f64 / signed_tokens as f64).abs();
+    let last_leg_price = 99_000.0 / 2_000.0;
     assert!(
-        (price - net_ratio).abs() < f64::EPSILON,
+        (price - last_leg_price).abs() < f64::EPSILON,
-        "single-direction transactions must be unaffected: {price} vs {net_ratio}"
+        "close should be last leg's price: {price} vs {last_leg_price}"
     );
 }
 
@@ -228,8 +227,12 @@ async fn test_exactly_cancelling_legs_still_price() {
         .unwrap()
         .expect("zero-net transaction must still price from its legs");
 
-    // 20_000 gross sats over 2_000 gross tokens, between the 9 and 11 leg prices.
+    // Per-leg pricing: the close is the last leg's price.
-    assert!((price - 10.0).abs() < f64::EPSILON, "got {price}");
+    // Last leg: (11_000, -1_000) = 11,000 / 1,000 = 11.0
+    assert!(
+        (price - 11.0).abs() < f64::EPSILON,
+        "close should be last leg's price: got {price}"
+    );
 }
 
 /// Legs that move no tokens cannot produce a price (division by zero volume).