feat(ml-alpha): tick-rule signed trade-flow inference at L1 (C16)

Replaces the placeholder `trade_signed_vol = trade_count_delta` (always
non-negative, sign-neutral) with a proper Databento-standard tick-rule
inference applied to L1 size + price deltas across consecutive MBP-10
snapshots:

  • ask_px[0] unchanged AND ask_sz[0] decreased → aggressive buys
    consumed ask depth; add (prev.ask_sz − cur.ask_sz).
  • bid_px[0] unchanged AND bid_sz[0] decreased → aggressive sells hit
    the bid; subtract (prev.bid_sz − cur.bid_sz).
  • ask_px[0] moved up → previous best ask cleared; add prev.ask_sz.
  • bid_px[0] moved down → previous best bid hit; subtract prev.bid_sz.
  • Pure cancellations (size shrank but price moved AWAY from us) =
    ambiguous; ignore.

Convention matches `Mbp10RawInput::trade_signed_vol`: positive =
buyer-initiated, negative = seller-initiated. This is a LOWER-BOUND
estimator — won't catch trades that cleared multiple levels (those
manifest only via deeper-level deltas) or trades against hidden /
off-book liquidity. Acceptable for v1 queue-decay signal; production
deployments can layer the trades-stream loader for ground-truth flow.

Wired into BacktestHarness::run() → sim.step_resting_orders(ts, vol)
so the queue-decay branch of resting_orders.cu finally fires with
non-zero input. Previously the harness passed 0.0 unconditionally,
which meant resting limits could only fill via the price-cross
marketability branch — same-price queue-decay was inert.

Six new unit tests cover each branch of the inference (pure cancel,
ask-shrank, bid-shrank, ask-px-up, bid-px-down, mixed both-sides).
34 ml-alpha lib tests + 33 ml-backtesting lib + 12 GPU fixtures + 3
fuzz still green.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

crates/ml-alpha/src/data/loader.rs

@@ -415,11 +415,7 @@ fn convert(
     }
     let prev_mid = mid_price_f32(prev);
     let trade_count = cur.trade_count.saturating_sub(prev.trade_count);
-    // `trade_signed_vol` isn't directly in Mbp10Snapshot; we'd compute it
-    // from the trades stream in production. For Phase A v1 use trade_count
-    // as a proxy (sign neutral); the snap_feature kernel reads it but the
-    // production pipeline will replace this with the trades-loader output.
-    let trade_signed_vol = (cur.trade_count as i64 - prev.trade_count as i64) as f32;
+    let trade_signed_vol = infer_signed_trade_flow(cur, prev);
     Mbp10RawInput {
         bid_px,
         bid_sz,
@@ -434,6 +430,125 @@ fn convert(
     }
 }
 
+/// Infer signed trade flow at L1 from MBP-10 snapshot deltas without
+/// the separate trades stream. Convention matches `Mbp10RawInput::
+/// trade_signed_vol`: positive = buyer-initiated (aggressive cross of
+/// the ask), negative = seller-initiated (aggressive hit of the bid).
+///
+/// Heuristic (Databento-standard tick rule applied to L1):
+///   * ask_px[0] unchanged AND ask_sz[0] decreased
+///     → aggressive buys consumed ask depth; add (prev.ask_sz - cur.ask_sz).
+///   * bid_px[0] unchanged AND bid_sz[0] decreased
+///     → aggressive sells consumed bid depth; subtract (prev.bid_sz - cur.bid_sz).
+///   * ask_px[0] moved up (best ask cleared completely)
+///     → infer full prev.ask_sz lifted; add prev.ask_sz.
+///   * bid_px[0] moved down (best bid cleared completely)
+///     → infer full prev.bid_sz hit; subtract prev.bid_sz.
+///   * Pure cancellations (size shrank but price moved away from us)
+///     ambiguous; ignore.
+///
+/// This is a lower-bound estimator — won't catch trades that crossed
+/// multiple levels (those show up only via the deeper-level deltas)
+/// nor trades against hidden / off-book liquidity. Acceptable for v1
+/// queue-decay signal; production deployments should layer in the
+/// trades-stream loader for ground-truth signed flow.
+fn infer_signed_trade_flow(cur: &Mbp10Snapshot, prev: &Mbp10Snapshot) -> f32 {
+    let cur_l1 = cur.levels.first().copied().unwrap_or_else(BidAskPair::empty);
+    let prev_l1 = prev.levels.first().copied().unwrap_or_else(BidAskPair::empty);
+
+    let cur_bid_px = BidAskPair::price_to_f64(cur_l1.bid_px) as f32;
+    let cur_bid_sz = cur_l1.bid_sz as f32;
+    let cur_ask_px = BidAskPair::price_to_f64(cur_l1.ask_px) as f32;
+    let cur_ask_sz = cur_l1.ask_sz as f32;
+    let prev_bid_px = BidAskPair::price_to_f64(prev_l1.bid_px) as f32;
+    let prev_bid_sz = prev_l1.bid_sz as f32;
+    let prev_ask_px = BidAskPair::price_to_f64(prev_l1.ask_px) as f32;
+    let prev_ask_sz = prev_l1.ask_sz as f32;
+
+    let mut buyer_init = 0.0_f32;
+    let mut seller_init = 0.0_f32;
+
+    // Same-price size decrement = aggressive cross at that price.
+    if (cur_ask_px - prev_ask_px).abs() < f32::EPSILON && cur_ask_sz < prev_ask_sz {
+        buyer_init += prev_ask_sz - cur_ask_sz;
+    }
+    if (cur_bid_px - prev_bid_px).abs() < f32::EPSILON && cur_bid_sz < prev_bid_sz {
+        seller_init += prev_bid_sz - cur_bid_sz;
+    }
+    // Best price moved away — infer the previous level fully cleared.
+    if cur_ask_px > prev_ask_px && prev_ask_sz > 0.0 {
+        buyer_init += prev_ask_sz;
+    }
+    if cur_bid_px < prev_bid_px && prev_bid_sz > 0.0 {
+        seller_init += prev_bid_sz;
+    }
+
+    buyer_init - seller_init
+}
+
+#[cfg(test)]
+mod trade_flow_tests {
+    use super::*;
+
+    fn snap_with_l1(ts: u64, bid_px: f64, bid_sz: u32, ask_px: f64, ask_sz: u32, trade_count: u32) -> Mbp10Snapshot {
+        let mut levels = vec![BidAskPair::empty(); 10];
+        levels[0] = BidAskPair {
+            bid_px: BidAskPair::price_from_f64(bid_px),
+            bid_sz,
+            bid_ct: 1,
+            ask_px: BidAskPair::price_from_f64(ask_px),
+            ask_sz,
+            ask_ct: 1,
+        };
+        Mbp10Snapshot::new("ES.FUT".into(), ts, levels, 0, trade_count)
+    }
+
+    #[test]
+    fn pure_cancel_no_size_change_reports_zero_flow() {
+        let prev = snap_with_l1(0, 5500.00, 100, 5500.25, 100, 0);
+        let cur  = snap_with_l1(1, 5500.00, 100, 5500.25, 100, 0);
+        assert_eq!(infer_signed_trade_flow(&cur, &prev), 0.0);
+    }
+
+    #[test]
+    fn ask_size_shrank_unchanged_price_reports_positive_buyer_flow() {
+        let prev = snap_with_l1(0, 5500.00, 100, 5500.25, 100, 0);
+        let cur  = snap_with_l1(1, 5500.00, 100, 5500.25,  60, 5);
+        // 100 → 60 = aggressive buys consumed 40 lots at the ask.
+        assert_eq!(infer_signed_trade_flow(&cur, &prev), 40.0);
+    }
+
+    #[test]
+    fn bid_size_shrank_unchanged_price_reports_negative_seller_flow() {
+        let prev = snap_with_l1(0, 5500.00, 100, 5500.25, 100, 0);
+        let cur  = snap_with_l1(1, 5500.00,  70, 5500.25, 100, 3);
+        assert_eq!(infer_signed_trade_flow(&cur, &prev), -30.0);
+    }
+
+    #[test]
+    fn ask_price_moved_up_reports_full_prev_ask_as_buyer_flow() {
+        // best ask cleared — infer prev.ask_sz fully lifted.
+        let prev = snap_with_l1(0, 5500.00, 100, 5500.25,  50, 0);
+        let cur  = snap_with_l1(1, 5500.00, 100, 5500.50, 100, 1);
+        assert_eq!(infer_signed_trade_flow(&cur, &prev), 50.0);
+    }
+
+    #[test]
+    fn bid_price_moved_down_reports_full_prev_bid_as_seller_flow() {
+        let prev = snap_with_l1(0, 5500.00,  80, 5500.25, 100, 0);
+        let cur  = snap_with_l1(1, 5499.75, 100, 5500.25, 100, 1);
+        assert_eq!(infer_signed_trade_flow(&cur, &prev), -80.0);
+    }
+
+    #[test]
+    fn mixed_buyer_and_seller_within_same_snapshot_nets_correctly() {
+        let prev = snap_with_l1(0, 5500.00, 100, 5500.25, 100, 0);
+        // ask shrank by 25 (aggressive buys), bid shrank by 10 (aggressive sells).
+        let cur  = snap_with_l1(1, 5500.00,  90, 5500.25,  75, 3);
+        assert_eq!(infer_signed_trade_flow(&cur, &prev), 15.0);
+    }
+}
+
 #[cfg(test)]
 mod inference_mode_tests {
     use super::*;

crates/ml-backtesting/src/harness.rs

@@ -124,10 +124,10 @@ impl BacktestHarness {
 
             // Process resting orders (in-flight promotion, queue decay,
             // marketability check, stop triggers, OCO). Trade-flow signal
-            // is currently unavailable from Mbp10RawInput (deferred to a
-            // trades-feed integration); pass 0.0 so only the price-cross
-            // marketability branch fires.
-            self.sim.step_resting_orders(raw.ts_ns, 0.0)?;
+            // comes from the loader's L1-delta tick-rule inference (see
+            // ml_alpha::data::loader::infer_signed_trade_flow). Positive =
+            // buyer-initiated, negative = seller-initiated.
+            self.sim.step_resting_orders(raw.ts_ns, raw.trade_signed_vol)?;
 
             // At decision-stride boundaries: run trunk inference + sim decision.
             if self.event_count % stride == 0 {