Deepbook Indexer Trades Endpoint Update (MystenLabs#20998)

## Description 

1. taker_balance_manager_id and maker_balance_manager_id included with
each trade
2. taker_order_id and maker_order_id included with each trade

Optional Filters:
1. limit: defaults to 1 (number of trades to be included in result).
Most recent trades are always included first.
2. taker_balance_manager_id: defaults to no filter
3. maker_balance_manager_id: defaults to no filter
4. start_time: unix timestamp in seconds, defaults to 24hrs before
end_time
5. end_time: unix timestamp in seconds, defaults to current time

Sample query:

`trades/SUI_USDC?start_time=1738093405&end_time=1738096485&limit=2&maker_balance_manager_id=0x344c2734b1d211bd15212bfb7847c66a3b18803f3f5ab00f5ff6f87b6fe6d27d&taker_balance_manager_id=0x47dcbbc8561fe3d52198336855f0983878152a12524749e054357ac2e3573d58`

Sample Response: 

`[{"base_volume":405.9,"type":"sell","maker_order_id":"68160737799100866923792791","taker_order_id":"170141183460537392451039660509112362617","quote_volume":1499.8005,"maker_balance_manager_id":"0x344c2734b1d211bd15212bfb7847c66a3b18803f3f5ab00f5ff6f87b6fe6d27d","trade_id":"136321457151457660152049680","price":3.695,"timestamp":1738096392913,"taker_balance_manager_id":"0x47dcbbc8561fe3d52198336855f0983878152a12524749e054357ac2e3573d58"},{"trade_id":"137612729236617328765169261","taker_balance_manager_id":"0x47dcbbc8561fe3d52198336855f0983878152a12524749e054357ac2e3573d58","maker_order_id":"68806373841680701230353053","type":"sell","maker_balance_manager_id":"0x344c2734b1d211bd15212bfb7847c66a3b18803f3f5ab00f5ff6f87b6fe6d27d","quote_volume":1499.833,"timestamp":1738095493338,"taker_order_id":"170141183460538038087082240343418921936","price":3.73,"base_volume":402.1}]`

## Test plan 

How did you test the new or updated feature?

Tested endpoint locally

## Release notes

Check each box that your changes affect. If none of the boxes relate to
your changes, release notes aren't required.

For each box you select, include information after the relevant heading
that describes the impact of your changes that a user might notice and
any actions they must take to implement updates.

- [ ] Protocol: 
- [ ] Nodes (Validators and Full nodes): 
- [ ] gRPC:
- [ ] JSON-RPC: 
- [ ] GraphQL: 
- [ ] CLI: 
- [ ] Rust SDK:

crates/sui-deepbook-indexer/src/server.rs

@@ -753,6 +753,7 @@ async fn price_change_24h(
 
 async fn trades(
     Path(pool_name): Path<String>,
+    Query(params): Query<HashMap<String, String>>,
     State(state): State<PgDeepbookPersistent>,
 ) -> Result<Json<Vec<HashMap<String, Value>>>, DeepBookError> {
     // Fetch all pools to map names to IDs and decimals
@@ -768,14 +769,56 @@ async fn trades(
         .await
         .map_err(|_| DeepBookError::InternalError(format!("Pool '{}' not found", pool_name)))?;
 
+    // Parse start_time and end_time
+    let end_time = params
+        .get("end_time")
+        .and_then(|v| v.parse::<i64>().ok())
+        .map(|t| t * 1000) // Convert to milliseconds
+        .unwrap_or_else(|| {
+            SystemTime::now()
+                .duration_since(UNIX_EPOCH)
+                .unwrap()
+                .as_millis() as i64
+        });
+
+    let start_time = params
+        .get("start_time")
+        .and_then(|v| v.parse::<i64>().ok())
+        .map(|t| t * 1000) // Convert to milliseconds
+        .unwrap_or_else(|| end_time - 24 * 60 * 60 * 1000);
+
+    // Parse limit (default to 1 if not provided)
+    let limit = params
+        .get("limit")
+        .and_then(|v| v.parse::<i64>().ok())
+        .unwrap_or(1);
+
+    // Parse optional filters for balance managers
+    let maker_balance_manager_filter = params.get("maker_balance_manager_id").cloned();
+    let taker_balance_manager_filter = params.get("taker_balance_manager_id").cloned();
+
     let (pool_id, base_decimals, quote_decimals) = pool_data;
     let base_decimals = base_decimals as u8;
     let quote_decimals = quote_decimals as u8;
 
-    // Fetch the last trade for the pool from the order_fills table
-    let last_trade = schema::order_fills::table
+    // Build the query dynamically
+    let mut query = schema::order_fills::table
         .filter(schema::order_fills::pool_id.eq(pool_id))
-        .order_by(schema::order_fills::checkpoint_timestamp_ms.desc())
+        .filter(schema::order_fills::checkpoint_timestamp_ms.between(start_time, end_time))
+        .into_boxed();
+
+    // Apply optional filters if parameters are provided
+    if let Some(maker_id) = maker_balance_manager_filter {
+        query = query.filter(schema::order_fills::maker_balance_manager_id.eq(maker_id));
+    }
+    if let Some(taker_id) = taker_balance_manager_filter {
+        query = query.filter(schema::order_fills::taker_balance_manager_id.eq(taker_id));
+    }
+
+    // Fetch latest trades (sorted by timestamp in descending order) within the time range, applying the limit
+    let trades = query
+        .order_by(schema::order_fills::checkpoint_timestamp_ms.desc()) // Ensures latest trades come first
+        .limit(limit) // Apply limit to get the most recent trades
         .select((
             schema::order_fills::maker_order_id,
             schema::order_fills::taker_order_id,
@@ -784,52 +827,73 @@ async fn trades(
             schema::order_fills::quote_quantity,
             schema::order_fills::checkpoint_timestamp_ms,
             schema::order_fills::taker_is_bid,
+            schema::order_fills::maker_balance_manager_id,
+            schema::order_fills::taker_balance_manager_id,
         ))
-        .first::<(String, String, i64, i64, i64, i64, bool)>(connection)
+        .load::<(String, String, i64, i64, i64, i64, bool, String, String)>(connection)
         .await
         .map_err(|_| {
-            DeepBookError::InternalError(format!("No trades found for pool '{}'", pool_name))
+            DeepBookError::InternalError(format!(
+                "No trades found for pool '{}' in the specified time range",
+                pool_name
+            ))
         })?;
 
-    let (
-        maker_order_id,
-        taker_order_id,
-        price,
-        base_quantity,
-        quote_quantity,
-        timestamp,
-        taker_is_bid,
-    ) = last_trade;
-
-    // Calculate the `trade_id` using the external function
-    let trade_id = calculate_trade_id(&maker_order_id, &taker_order_id)?;
-
     // Conversion factors for decimals
     let base_factor = 10u64.pow(base_decimals as u32);
     let quote_factor = 10u64.pow(quote_decimals as u32);
     let price_factor = 10u64.pow((9 - base_decimals + quote_decimals) as u32);
-    let trade_type = if taker_is_bid { "buy" } else { "sell" };
 
-    // Prepare the trade data
-    let trade = HashMap::from([
-        ("trade_id".to_string(), Value::from(trade_id.to_string())), // Computed from `maker_id` and `taker_id`
-        (
-            "price".to_string(),
-            Value::from(price as f64 / price_factor as f64),
-        ),
-        (
-            "base_volume".to_string(),
-            Value::from(base_quantity as f64 / base_factor as f64),
-        ),
-        (
-            "quote_volume".to_string(),
-            Value::from(quote_quantity as f64 / quote_factor as f64),
-        ),
-        ("timestamp".to_string(), Value::from(timestamp as u64)),
-        ("type".to_string(), Value::from(trade_type)), // Trade type (buy/sell)
-    ]);
-
-    Ok(Json(vec![trade]))
+    // Map trades to JSON format
+    let trade_data: Vec<HashMap<String, Value>> = trades
+        .into_iter()
+        .map(
+            |(
+                maker_order_id,
+                taker_order_id,
+                price,
+                base_quantity,
+                quote_quantity,
+                timestamp,
+                taker_is_bid,
+                maker_balance_manager_id,
+                taker_balance_manager_id,
+            )| {
+                let trade_id = calculate_trade_id(&maker_order_id, &taker_order_id).unwrap_or(0);
+                let trade_type = if taker_is_bid { "buy" } else { "sell" };
+
+                HashMap::from([
+                    ("trade_id".to_string(), Value::from(trade_id.to_string())),
+                    ("maker_order_id".to_string(), Value::from(maker_order_id)),
+                    ("taker_order_id".to_string(), Value::from(taker_order_id)),
+                    (
+                        "maker_balance_manager_id".to_string(),
+                        Value::from(maker_balance_manager_id),
+                    ),
+                    (
+                        "taker_balance_manager_id".to_string(),
+                        Value::from(taker_balance_manager_id),
+                    ),
+                    (
+                        "price".to_string(),
+                        Value::from(price as f64 / price_factor as f64),
+                    ),
+                    (
+                        "base_volume".to_string(),
+                        Value::from(base_quantity as f64 / base_factor as f64),
+                    ),
+                    (
+                        "quote_volume".to_string(),
+                        Value::from(quote_quantity as f64 / quote_factor as f64),
+                    ),
+                    ("timestamp".to_string(), Value::from(timestamp as u64)),
+                    ("type".to_string(), Value::from(trade_type)),
+                ])
+            },
+        )
+        .collect();
+
+    Ok(Json(trade_data))
 }
 
 async fn trade_count(