hypercall_db_diesel/

analytics.rs

//! Analytics trait implementations for DieselDb.
//!
//! Implements `AnalyticsReader` and `AnalyticsWriter` from `hypercall-db`,
//! covering historical PnL, theos, BBO snapshots, vol surfaces, settlement
//! payouts, instrument existence checks, and trade/fill/order queries.

use std::collections::{HashMap, HashSet};

use anyhow::Result;
use diesel::prelude::*;
use diesel::upsert::excluded;
use diesel_async::{AsyncConnection, RunQueryDsl};
use rust_decimal::Decimal;
use struct_convert::Convert;

use crate::diesel_db::DieselDb;
use crate::models::{
    BboSnapshot, HistoricalTheoSnapshot, NewBboSnapshot, NewHistoricalPnlSnapshot,
    NewHistoricalTheoSnapshot, NewVolSurfaceSnapshotRow, SettlementPayout, VolSurfaceSnapshotRow,
};
use crate::schema;
use hypercall_db::{
    AnalyticsReader, AnalyticsWriter, BboReferenceData, BboSnapshotRecord, FillRecord,
    HistoricalPnlPoint, HistoricalTheoPoint, NewBboSnapshotInput, OrderRecord,
    SettlementPayoutRecord, TradeRecord, VolSurfaceSnapshot,
};
use hypercall_types::{
    WalletAddress, HISTORICAL_PNL_INTERVAL_1D_MS, HISTORICAL_PNL_INTERVAL_1H_MS,
    HISTORICAL_PNL_INTERVAL_5M_MS, HISTORICAL_THEO_INTERVAL_1D_MS, HISTORICAL_THEO_INTERVAL_1H_MS,
    HISTORICAL_THEO_INTERVAL_5M_MS,
};

use crate::models::NewSettlementPayoutSeen;

// ---- Private row types for raw SQL queries ----

#[derive(QueryableByName, Debug)]
struct BboReferenceRow {
    #[diesel(sql_type = diesel::sql_types::Text)]
    pub symbol: String,
    #[diesel(sql_type = diesel::sql_types::Numeric)]
    pub reference_ask: Decimal,
    #[diesel(sql_type = diesel::sql_types::BigInt)]
    pub reference_ts: i64,
    #[diesel(sql_type = diesel::sql_types::Bool)]
    pub used_earliest_fallback: bool,
}

#[derive(QueryableByName, Debug, Convert)]
#[convert(from_on = "TradeRecord")]
struct TradeRow {
    #[diesel(sql_type = diesel::sql_types::BigInt)]
    pub trade_id: i64,
    #[diesel(sql_type = diesel::sql_types::Text)]
    pub symbol: String,
    #[diesel(sql_type = diesel::sql_types::Numeric)]
    pub price: Decimal,
    #[diesel(sql_type = diesel::sql_types::Numeric)]
    pub size: Decimal,
    #[diesel(sql_type = diesel::sql_types::Bytea)]
    pub maker_address: WalletAddress,
    #[diesel(sql_type = diesel::sql_types::Bytea)]
    pub taker_address: WalletAddress,
    #[diesel(sql_type = diesel::sql_types::Numeric)]
    pub maker_fee: Decimal,
    #[diesel(sql_type = diesel::sql_types::Numeric)]
    pub taker_fee: Decimal,
    #[diesel(sql_type = diesel::sql_types::BigInt)]
    pub timestamp: i64,
    #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::Timestamp>)]
    #[convert_field(
        custom_fn = ".map(|dt| chrono::DateTime::<chrono::Utc>::from_naive_utc_and_offset(dt, chrono::Utc))"
    )]
    pub created_at: Option<chrono::NaiveDateTime>,
}

#[derive(QueryableByName, Debug, Convert)]
#[convert(from_on = "FillRecord")]
struct FillRow {
    #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::BigInt>)]
    pub fill_id: Option<i64>,
    #[diesel(sql_type = diesel::sql_types::BigInt)]
    pub trade_id: i64,
    #[diesel(sql_type = diesel::sql_types::Bytea)]
    pub wallet_address: WalletAddress,
    #[diesel(sql_type = diesel::sql_types::Text)]
    pub symbol: String,
    #[diesel(sql_type = diesel::sql_types::Numeric)]
    pub price: Decimal,
    #[diesel(sql_type = diesel::sql_types::Numeric)]
    pub size: Decimal,
    #[diesel(sql_type = diesel::sql_types::Numeric)]
    pub fee: Decimal,
    #[diesel(sql_type = diesel::sql_types::Bool)]
    pub is_taker: bool,
    #[diesel(sql_type = diesel::sql_types::BigInt)]
    pub timestamp: i64,
    #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::Timestamp>)]
    #[convert_field(
        custom_fn = ".map(|dt| chrono::DateTime::<chrono::Utc>::from_naive_utc_and_offset(dt, chrono::Utc))"
    )]
    pub created_at: Option<chrono::NaiveDateTime>,
    #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::Bytea>)]
    pub builder_code_address: Option<WalletAddress>,
    #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::Numeric>)]
    pub builder_code_fee: Option<Decimal>,
    #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::Numeric>)]
    pub realized_pnl: Option<Decimal>,
    #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::Numeric>)]
    pub underlying_notional: Option<Decimal>,
}

#[derive(QueryableByName, Debug, Convert)]
#[convert(from_on = "OrderRecord")]
struct OrderRow {
    #[diesel(sql_type = diesel::sql_types::BigInt)]
    pub order_id: i64,
    #[diesel(sql_type = diesel::sql_types::Bytea)]
    pub wallet_address: WalletAddress,
    #[diesel(sql_type = diesel::sql_types::Text)]
    pub symbol: String,
    #[diesel(sql_type = diesel::sql_types::Text)]
    pub side: String,
    #[diesel(sql_type = diesel::sql_types::Numeric)]
    pub price: Decimal,
    #[diesel(sql_type = diesel::sql_types::Numeric)]
    pub size: Decimal,
    #[diesel(sql_type = diesel::sql_types::Text)]
    pub tif: String,
    #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::Text>)]
    pub client_id: Option<String>,
    #[diesel(sql_type = diesel::sql_types::Bool)]
    pub is_perp: bool,
    #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::Text>)]
    pub underlying: Option<String>,
    #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::Bool>)]
    pub reduce_only: Option<bool>,
    #[diesel(sql_type = diesel::sql_types::Text)]
    pub status: String,
    #[diesel(sql_type = diesel::sql_types::BigInt)]
    #[convert_field(rename = "timestamp")]
    pub created_at: i64,
    #[diesel(sql_type = diesel::sql_types::Timestamp)]
    #[convert_field(
        rename = "created_at",
        custom_fn = "Some(chrono::DateTime::<chrono::Utc>::from_naive_utc_and_offset(this.updated_at, chrono::Utc))"
    )]
    pub updated_at: chrono::NaiveDateTime,
    #[diesel(sql_type = diesel::sql_types::Numeric)]
    pub filled_size: Decimal,
    #[diesel(sql_type = diesel::sql_types::Bool)]
    pub mmp_enabled: bool,
}

fn normalize_materialized_order_status_filter(status: &str) -> String {
    match status.trim().to_ascii_lowercase().as_str() {
        "open" => "OPEN".to_string(),
        "acked" => "ACKED".to_string(),
        "partially_filled" | "partially-filled" => "PARTIALLY_FILLED".to_string(),
        "filled" => "FILLED".to_string(),
        "canceled" | "cancelled" => "CANCELED".to_string(),
        "rejected" => "REJECTED".to_string(),
        other => other.to_ascii_uppercase().replace('-', "_"),
    }
}

// =============================================================================
// AnalyticsReader implementation
// =============================================================================

#[async_trait::async_trait]
impl AnalyticsReader for DieselDb {
    async fn get_all_trades(&self, limit: usize, offset: usize) -> Result<Vec<TradeRecord>> {
        let mut conn = self.get_conn().await?;

        let trades = diesel::sql_query(
            "SELECT trade_id, symbol, price, size,
                    maker_address, taker_address, maker_fee,
                    taker_fee, timestamp, created_at
             FROM trades
             ORDER BY timestamp DESC
             LIMIT $1 OFFSET $2",
        )
        .bind::<diesel::sql_types::Integer, _>(limit as i32)
        .bind::<diesel::sql_types::Integer, _>(offset as i32)
        .load::<TradeRow>(&mut conn)
        .await?;

        Ok(trades.into_iter().map(Into::into).collect())
    }

    async fn get_trades_by_option(
        &self,
        option_id: &str,
        limit: usize,
    ) -> Result<Vec<TradeRecord>> {
        let mut conn = self.get_conn().await?;

        let trades = diesel::sql_query(
            "SELECT trade_id, symbol, price, size,
                    maker_address, taker_address, maker_fee,
                    taker_fee, timestamp, created_at
             FROM trades
             WHERE symbol = $1
             ORDER BY timestamp DESC
             LIMIT $2",
        )
        .bind::<diesel::sql_types::Text, _>(option_id)
        .bind::<diesel::sql_types::Integer, _>(limit as i32)
        .load::<TradeRow>(&mut conn)
        .await?;

        Ok(trades.into_iter().map(Into::into).collect())
    }

    async fn get_trades_by_underlying(
        &self,
        underlying: &str,
        limit: usize,
        offset: usize,
    ) -> Result<Vec<TradeRecord>> {
        let mut conn = self.get_conn().await?;

        let trades = diesel::sql_query(
            "SELECT t.trade_id, t.symbol, t.price, t.size,
                    t.maker_address, t.taker_address, t.maker_fee,
                    t.taker_fee, t.timestamp, t.created_at
             FROM trades t
             INNER JOIN instruments i ON t.symbol = i.id
             WHERE i.underlying = $1
             ORDER BY t.timestamp DESC
             LIMIT $2 OFFSET $3",
        )
        .bind::<diesel::sql_types::Text, _>(underlying)
        .bind::<diesel::sql_types::Integer, _>(limit as i32)
        .bind::<diesel::sql_types::Integer, _>(offset as i32)
        .load::<TradeRow>(&mut conn)
        .await?;

        Ok(trades.into_iter().map(Into::into).collect())
    }

    async fn get_trades_by_account(
        &self,
        account: &WalletAddress,
        limit: usize,
        offset: usize,
    ) -> Result<Vec<TradeRecord>> {
        let mut conn = self.get_conn().await?;

        let trades = diesel::sql_query(
            "SELECT trade_id, symbol, price, size,
                    maker_address, taker_address, maker_fee,
                    taker_fee, timestamp, created_at
             FROM trades
             WHERE maker_address = $1 OR taker_address = $1
             ORDER BY timestamp DESC
             LIMIT $2 OFFSET $3",
        )
        .bind::<diesel::sql_types::Bytea, _>(account)
        .bind::<diesel::sql_types::Integer, _>(limit as i32)
        .bind::<diesel::sql_types::Integer, _>(offset as i32)
        .load::<TradeRow>(&mut conn)
        .await?;

        Ok(trades.into_iter().map(Into::into).collect())
    }

    async fn get_trades_for_symbol_in_range(
        &self,
        symbol: &str,
        start_time_ms: i64,
        end_time_ms: i64,
    ) -> Result<Vec<TradeRecord>> {
        let mut conn = self.get_conn().await?;

        let trades = diesel::sql_query(
            "SELECT trade_id, symbol, price, size,
                    maker_address, taker_address, maker_fee,
                    taker_fee, timestamp, created_at
             FROM trades
             WHERE symbol = $1 AND timestamp >= $2 AND timestamp <= $3
             ORDER BY timestamp ASC",
        )
        .bind::<diesel::sql_types::Text, _>(symbol)
        .bind::<diesel::sql_types::BigInt, _>(start_time_ms)
        .bind::<diesel::sql_types::BigInt, _>(end_time_ms)
        .load::<TradeRow>(&mut conn)
        .await?;

        Ok(trades.into_iter().map(Into::into).collect())
    }

    async fn get_fills_by_account(
        &self,
        account: &WalletAddress,
        limit: usize,
        offset: usize,
    ) -> Result<Vec<FillRecord>> {
        let mut conn = self.get_conn().await?;

        let fills = diesel::sql_query(
            "SELECT fill_id, trade_id, wallet_address, symbol, price, size, fee, is_taker, timestamp, created_at, builder_code_address, builder_code_fee, realized_pnl, underlying_notional
             FROM fills
             WHERE wallet_address = $1
             ORDER BY timestamp DESC
             LIMIT $2 OFFSET $3",
        )
        .bind::<diesel::sql_types::Binary, _>(account)
        .bind::<diesel::sql_types::Integer, _>(limit as i32)
        .bind::<diesel::sql_types::Integer, _>(offset as i32)
        .load::<FillRow>(&mut conn)
        .await?;

        Ok(fills.into_iter().map(Into::into).collect())
    }

    async fn get_orders_by_account(
        &self,
        account: &WalletAddress,
        status: Option<&str>,
        limit: usize,
        offset: usize,
    ) -> Result<Vec<OrderRecord>> {
        let mut conn = self.get_conn().await?;

        let orders = if let Some(status_filter) = status {
            let normalized_status = normalize_materialized_order_status_filter(status_filter);
            diesel::sql_query(
                "SELECT
                    oi.order_id,
                    oi.wallet_address,
                    oi.symbol,
                    oi.side,
                    oi.price,
                    oi.size,
                    oi.tif,
                    oi.client_id,
                    oi.is_perp,
                    oi.underlying,
                    oi.reduce_only,
                    oi.status,
                    oi.timestamp as created_at,
                    oi.updated_at,
                    oi.filled_size,
                    oi.mmp_enabled
                FROM order_infos oi
                WHERE oi.wallet_address = $1 AND oi.status = $2
                ORDER BY oi.timestamp DESC
                LIMIT $3 OFFSET $4",
            )
            .bind::<diesel::sql_types::Binary, _>(account)
            .bind::<diesel::sql_types::Text, _>(normalized_status)
            .bind::<diesel::sql_types::Integer, _>(limit as i32)
            .bind::<diesel::sql_types::Integer, _>(offset as i32)
            .load::<OrderRow>(&mut conn)
            .await?
        } else {
            diesel::sql_query(
                "SELECT
                    oi.order_id,
                    oi.wallet_address,
                    oi.symbol,
                    oi.side,
                    oi.price,
                    oi.size,
                    oi.tif,
                    oi.client_id,
                    oi.is_perp,
                    oi.underlying,
                    oi.reduce_only,
                    oi.status,
                    oi.timestamp as created_at,
                    oi.updated_at,
                    oi.filled_size,
                    oi.mmp_enabled
                FROM order_infos oi
                WHERE oi.wallet_address = $1
                ORDER BY oi.timestamp DESC
                LIMIT $2 OFFSET $3",
            )
            .bind::<diesel::sql_types::Binary, _>(account)
            .bind::<diesel::sql_types::Integer, _>(limit as i32)
            .bind::<diesel::sql_types::Integer, _>(offset as i32)
            .load::<OrderRow>(&mut conn)
            .await?
        };

        Ok(orders.into_iter().map(Into::into).collect())
    }

    async fn trade_history_exists_for_symbol(&self, symbol: &str) -> Result<bool> {
        let mut conn = self.get_conn().await?;

        #[derive(QueryableByName, Debug)]
        struct ExistsRow {
            #[diesel(sql_type = diesel::sql_types::Bool)]
            exists: bool,
        }

        let row =
            diesel::sql_query("SELECT EXISTS(SELECT 1 FROM trades WHERE symbol = $1) AS exists")
                .bind::<diesel::sql_types::Text, _>(symbol)
                .get_result::<ExistsRow>(&mut conn)
                .await?;

        Ok(row.exists)
    }

    async fn historical_theo_history_exists_for_symbol(&self, symbol: &str) -> Result<bool> {
        use schema::historical_theo_snapshots::dsl as ht;

        let mut conn = self.get_conn().await?;

        let exists = diesel::select(diesel::dsl::exists(
            ht::historical_theo_snapshots.filter(ht::symbol.eq(symbol)),
        ))
        .get_result::<bool>(&mut conn)
        .await?;

        Ok(exists)
    }

    /// Historical equity point from DB, used by GET /historical-pnl.
    ///
    /// When `include_attribution` is false, the large `attribution` bytea is
    /// omitted from the SELECT.
    async fn get_historical_pnl(
        &self,
        wallet: &WalletAddress,
        interval_ms: i64,
        limit: usize,
        include_attribution: bool,
    ) -> Result<Vec<HistoricalPnlPoint>> {
        use schema::historical_pnl_snapshots::dsl as hp;

        let mut conn = self.get_conn().await?;

        let mut rows: Vec<HistoricalPnlPoint> = if include_attribution {
            hp::historical_pnl_snapshots
                .filter(hp::wallet_address.eq(wallet))
                .filter(hp::interval_ms.eq(interval_ms))
                .order(hp::timestamp_ms.desc())
                .limit(limit as i64)
                .select((hp::timestamp_ms, hp::total_equity, hp::attribution))
                .load::<(i64, Decimal, Option<Vec<u8>>)>(&mut conn)
                .await?
                .into_iter()
                .map(
                    |(timestamp_ms, total_equity, attribution)| HistoricalPnlPoint {
                        timestamp_ms,
                        total_equity,
                        attribution,
                    },
                )
                .collect()
        } else {
            hp::historical_pnl_snapshots
                .filter(hp::wallet_address.eq(wallet))
                .filter(hp::interval_ms.eq(interval_ms))
                .order(hp::timestamp_ms.desc())
                .limit(limit as i64)
                .select((hp::timestamp_ms, hp::total_equity))
                .load::<(i64, Decimal)>(&mut conn)
                .await?
                .into_iter()
                .map(|(timestamp_ms, total_equity)| HistoricalPnlPoint {
                    timestamp_ms,
                    total_equity,
                    attribution: None,
                })
                .collect()
        };

        rows.reverse();
        Ok(rows)
    }

    async fn get_historical_theos(
        &self,
        symbol: &str,
        interval_ms: i64,
        limit: usize,
    ) -> Result<Vec<HistoricalTheoPoint>> {
        use schema::historical_theo_snapshots::dsl as ht;

        let mut conn = self.get_conn().await?;

        let mut rows = ht::historical_theo_snapshots
            .filter(ht::symbol.eq(symbol))
            .filter(ht::interval_ms.eq(interval_ms))
            .order(ht::timestamp_ms.desc())
            .limit(limit as i64)
            .load::<HistoricalTheoSnapshot>(&mut conn)
            .await?;

        rows.reverse();

        Ok(rows
            .into_iter()
            .map(|row| HistoricalTheoPoint {
                timestamp_ms: row.timestamp_ms,
                theoretical_price: row.theoretical_price,
            })
            .collect())
    }

    async fn get_historical_theos_batch(
        &self,
        symbols: &[String],
        interval_ms: i64,
        limit: usize,
    ) -> Result<HashMap<String, Vec<HistoricalTheoPoint>>> {
        use schema::historical_theo_snapshots::dsl as ht;

        if symbols.is_empty() {
            return Ok(HashMap::new());
        }

        let mut conn = self.get_conn().await?;

        let rows = ht::historical_theo_snapshots
            .filter(ht::symbol.eq_any(symbols))
            .filter(ht::interval_ms.eq(interval_ms))
            .order((ht::symbol.asc(), ht::timestamp_ms.desc()))
            .load::<HistoricalTheoSnapshot>(&mut conn)
            .await?;

        let mut result: HashMap<String, Vec<HistoricalTheoPoint>> = HashMap::new();
        for row in rows {
            let entry = result.entry(row.symbol.clone()).or_default();
            if entry.len() < limit {
                entry.push(HistoricalTheoPoint {
                    timestamp_ms: row.timestamp_ms,
                    theoretical_price: row.theoretical_price,
                });
            }
        }

        for points in result.values_mut() {
            points.reverse();
        }

        Ok(result)
    }

    async fn get_vol_surface_history(
        &self,
        underlying: &str,
        interval_ms: i64,
        limit: usize,
    ) -> Result<Vec<VolSurfaceSnapshot>> {
        use schema::vol_surface_snapshots::dsl as vs;

        let mut conn = self.get_conn().await?;

        let mut rows = vs::vol_surface_snapshots
            .filter(vs::underlying.eq(underlying))
            .filter(vs::interval_ms.eq(interval_ms))
            .order(vs::timestamp_ms.desc())
            .limit(limit as i64)
            .load::<VolSurfaceSnapshotRow>(&mut conn)
            .await?;

        rows.reverse();

        Ok(rows
            .into_iter()
            .map(|row| VolSurfaceSnapshot {
                id: row.id,
                underlying: row.underlying,
                interval_ms: row.interval_ms,
                timestamp_ms: row.timestamp_ms,
                surface_json: row.surface_json,
                created_at: row.created_at,
            })
            .collect())
    }

    async fn load_bbo_snapshots_since(&self, cutoff_ts: i64) -> Result<Vec<BboSnapshotRecord>> {
        use schema::bbo_snapshots::dsl as bbo;

        let mut conn = self.get_conn().await?;

        let rows = bbo::bbo_snapshots
            .filter(bbo::snapshot_ts.ge(cutoff_ts))
            .order((bbo::symbol.asc(), bbo::snapshot_ts.asc()))
            .load::<BboSnapshot>(&mut conn)
            .await?;

        Ok(rows
            .into_iter()
            .map(|row| BboSnapshotRecord {
                id: row.id,
                symbol: row.symbol,
                best_bid: row.best_bid,
                best_ask: row.best_ask,
                best_bid_size: row.best_bid_size,
                best_ask_size: row.best_ask_size,
                snapshot_ts: row.snapshot_ts,
                created_at: row.created_at,
            })
            .collect())
    }

    async fn get_bbo_reference_asks(
        &self,
        symbols: &[String],
        cutoff_ts: i64,
    ) -> Result<HashMap<String, BboReferenceData>> {
        if symbols.is_empty() {
            return Ok(HashMap::new());
        }

        let mut conn = self.get_conn().await?;

        let rows = diesel::sql_query(
            "WITH requested AS (
                SELECT UNNEST($1::text[]) AS symbol
            ),
            latest_before AS (
                SELECT DISTINCT ON (b.symbol) b.symbol, b.best_ask, b.snapshot_ts
                FROM bbo_snapshots b
                JOIN requested r ON r.symbol = b.symbol
                WHERE b.snapshot_ts <= $2
                ORDER BY b.symbol, b.snapshot_ts DESC
            ),
            earliest_any AS (
                SELECT DISTINCT ON (b.symbol) b.symbol, b.best_ask, b.snapshot_ts
                FROM bbo_snapshots b
                JOIN requested r ON r.symbol = b.symbol
                ORDER BY b.symbol, b.snapshot_ts ASC
            )
            SELECT r.symbol,
                   COALESCE(lb.best_ask, ea.best_ask) AS reference_ask,
                   COALESCE(lb.snapshot_ts, ea.snapshot_ts) AS reference_ts,
                   (lb.symbol IS NULL AND ea.symbol IS NOT NULL) AS used_earliest_fallback
            FROM requested r
            LEFT JOIN latest_before lb USING (symbol)
            LEFT JOIN earliest_any ea USING (symbol)
            WHERE COALESCE(lb.symbol, ea.symbol) IS NOT NULL",
        )
        .bind::<diesel::sql_types::Array<diesel::sql_types::Text>, _>(symbols)
        .bind::<diesel::sql_types::BigInt, _>(cutoff_ts)
        .load::<BboReferenceRow>(&mut conn)
        .await?;

        Ok(rows
            .into_iter()
            .map(|row| {
                (
                    row.symbol,
                    BboReferenceData {
                        reference_ask: row.reference_ask,
                        reference_ts: row.reference_ts,
                        used_earliest_fallback: row.used_earliest_fallback,
                    },
                )
            })
            .collect())
    }

    async fn get_settlement_payouts(
        &self,
        wallet: &WalletAddress,
        limit: i64,
        offset: i64,
        symbol: Option<&str>,
        ledger_applied: Option<bool>,
    ) -> Result<(Vec<SettlementPayoutRecord>, i64)> {
        use schema::settlement_payouts::dsl as sp;

        let mut conn = self.get_conn().await?;

        let mut count_query = sp::settlement_payouts
            .filter(sp::wallet.eq(wallet))
            .into_boxed();
        let mut rows_query = sp::settlement_payouts
            .filter(sp::wallet.eq(wallet))
            .into_boxed();

        if let Some(symbol_filter) = symbol {
            count_query = count_query.filter(sp::symbol.eq(symbol_filter));
            rows_query = rows_query.filter(sp::symbol.eq(symbol_filter));
        }

        if let Some(applied_filter) = ledger_applied {
            count_query = count_query.filter(sp::ledger_applied.eq(applied_filter));
            rows_query = rows_query.filter(sp::ledger_applied.eq(applied_filter));
        }

        let count: i64 = count_query.count().get_result(&mut conn).await?;
        let rows = rows_query
            .order(sp::id.desc())
            .limit(limit)
            .offset(offset)
            .load::<SettlementPayout>(&mut conn)
            .await?;

        Ok((rows.into_iter().map(Into::into).collect(), count))
    }

    async fn get_seen_settlement_payout_ids(
        &self,
        wallet: &WalletAddress,
        payout_ids: &[i64],
    ) -> Result<HashSet<i64>> {
        use schema::settlement_payout_seen::dsl as sps;

        if payout_ids.is_empty() {
            return Ok(HashSet::new());
        }

        let mut conn = self.get_conn().await?;

        let rows: Vec<i64> = sps::settlement_payout_seen
            .filter(sps::wallet.eq(wallet))
            .filter(sps::payout_id.eq_any(payout_ids))
            .select(sps::payout_id)
            .load::<i64>(&mut conn)
            .await?;

        Ok(rows.into_iter().collect())
    }

    async fn get_client_ids_by_order_ids(
        &self,
        order_ids: &[i64],
    ) -> Result<HashMap<i64, Option<String>>> {
        if order_ids.is_empty() {
            return Ok(HashMap::new());
        }
        let mut conn = self.get_conn().await?;
        let results = crate::schema::order_infos::table
            .filter(crate::schema::order_infos::order_id.eq_any(order_ids))
            .select((
                crate::schema::order_infos::order_id,
                crate::schema::order_infos::client_id,
            ))
            .load::<(i64, Option<String>)>(&mut conn)
            .await?;
        Ok(results.into_iter().collect())
    }

    async fn get_theo_marks_at_timestamp(
        &self,
        symbols: &[String],
        timestamp_ms: i64,
    ) -> Result<HashMap<String, Decimal>> {
        if symbols.is_empty() {
            return Ok(HashMap::new());
        }

        #[derive(QueryableByName)]
        struct TheoRow {
            #[diesel(sql_type = diesel::sql_types::Text)]
            symbol: String,
            #[diesel(sql_type = diesel::sql_types::Numeric)]
            theoretical_price: Decimal,
        }

        let mut conn = self.get_conn().await?;
        let rows = diesel::sql_query(
            "SELECT DISTINCT ON (symbol) symbol, theoretical_price
             FROM historical_theo_snapshots
             WHERE symbol = ANY($1) AND timestamp_ms <= $2
             ORDER BY symbol, timestamp_ms DESC, interval_ms ASC",
        )
        .bind::<diesel::sql_types::Array<diesel::sql_types::Text>, _>(symbols)
        .bind::<diesel::sql_types::BigInt, _>(timestamp_ms)
        .get_results::<TheoRow>(&mut conn)
        .await?;

        Ok(rows
            .into_iter()
            .map(|row| (row.symbol, row.theoretical_price))
            .collect())
    }

    async fn get_deposit_withdraw_events(
        &self,
        wallet: &WalletAddress,
    ) -> Result<Vec<hypercall_db::LedgerEventTimeDelta>> {
        #[derive(QueryableByName)]
        struct LedgerEventRow {
            #[diesel(sql_type = diesel::sql_types::BigInt)]
            event_ts_ms: i64,
            #[diesel(sql_type = diesel::sql_types::Numeric)]
            delta: Decimal,
        }

        let mut conn = self.get_conn().await?;
        let rows = diesel::sql_query(
            "SELECT event_ts_ms, delta
             FROM ledger_events
             WHERE wallet = $1 AND event_type IN ('deposit', 'withdraw')
             ORDER BY event_ts_ms ASC, id ASC",
        )
        .bind::<diesel::sql_types::Bytea, _>(wallet)
        .get_results::<LedgerEventRow>(&mut conn)
        .await?;

        Ok(rows
            .into_iter()
            .map(|r| hypercall_db::LedgerEventTimeDelta {
                event_ts_ms: r.event_ts_ms,
                delta: r.delta,
            })
            .collect())
    }

    async fn get_settled_pnl_by_symbol(
        &self,
        wallet: &WalletAddress,
        before_ts_ms: i64,
    ) -> Result<Vec<(String, Decimal)>> {
        #[derive(QueryableByName)]
        struct SettlementRow {
            #[diesel(sql_type = diesel::sql_types::Text)]
            reference_symbol: String,
            #[diesel(sql_type = diesel::sql_types::Numeric)]
            total_pnl: Decimal,
        }

        let mut conn = self.get_conn().await?;
        let rows = diesel::sql_query(
            "SELECT reference_symbol, SUM(delta) AS total_pnl
             FROM ledger_events
             WHERE wallet = $1
               AND event_ts_ms <= $2
               AND event_type IN (
                   'fill_premium',
                   'fill_realized_pnl',
                   'settlement_realized_pnl'
               )
               AND reference_symbol IS NOT NULL
             GROUP BY reference_symbol",
        )
        .bind::<diesel::sql_types::Bytea, _>(wallet)
        .bind::<diesel::sql_types::BigInt, _>(before_ts_ms)
        .get_results::<SettlementRow>(&mut conn)
        .await?;

        Ok(rows
            .into_iter()
            .map(|r| (r.reference_symbol, r.total_pnl))
            .collect())
    }

    async fn get_fills_since_timestamp(
        &self,
        cutoff_timestamp: i64,
    ) -> Result<Vec<hypercall_db::FillBackfillRecord>> {
        #[derive(QueryableByName)]
        struct FillBackfillRow {
            #[diesel(sql_type = diesel::sql_types::Text)]
            symbol: String,
            #[diesel(sql_type = diesel::sql_types::Numeric)]
            price: Decimal,
            #[diesel(sql_type = diesel::sql_types::Numeric)]
            size: Decimal,
            #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::Numeric>)]
            underlying_notional: Option<Decimal>,
            #[diesel(sql_type = diesel::sql_types::BigInt)]
            timestamp: i64,
        }

        let mut conn = self.get_conn().await?;
        let rows = diesel::sql_query(
            "SELECT symbol, price, size, underlying_notional, timestamp
             FROM fills
             WHERE timestamp >= $1
               AND is_taker = true
             ORDER BY timestamp ASC",
        )
        .bind::<diesel::sql_types::BigInt, _>(cutoff_timestamp)
        .get_results::<FillBackfillRow>(&mut conn)
        .await?;

        Ok(rows
            .into_iter()
            .map(|r| hypercall_db::FillBackfillRecord {
                symbol: r.symbol,
                price: r.price,
                size: r.size,
                underlying_notional: r.underlying_notional,
                timestamp: r.timestamp,
            })
            .collect())
    }
}

// =============================================================================
// AnalyticsWriter implementation
// =============================================================================

#[async_trait::async_trait]
impl AnalyticsWriter for DieselDb {
    async fn upsert_historical_pnl_batch(
        &self,
        interval_ms: i64,
        timestamp_ms: i64,
        snapshots: &[(WalletAddress, Decimal, Option<Vec<u8>>)],
        max_periods: i64,
    ) -> Result<usize> {
        use schema::historical_pnl_snapshots::dsl as hp;

        if snapshots.is_empty() {
            return Ok(0);
        }
        if max_periods <= 0 {
            anyhow::bail!("max_periods must be positive, got {}", max_periods);
        }
        if ![
            HISTORICAL_PNL_INTERVAL_5M_MS,
            HISTORICAL_PNL_INTERVAL_1H_MS,
            HISTORICAL_PNL_INTERVAL_1D_MS,
        ]
        .contains(&interval_ms)
        {
            anyhow::bail!(
                "Invalid interval_ms {}: must be one of the known intervals (5m, 1h, 1d)",
                interval_ms
            );
        }

        let mut conn = self.get_conn().await?;

        let new_rows: Vec<NewHistoricalPnlSnapshot> = snapshots
            .iter()
            .map(|(wallet, equity, attr)| NewHistoricalPnlSnapshot {
                wallet_address: *wallet,
                interval_ms,
                timestamp_ms,
                total_equity: *equity,
                attribution: attr.clone(),
            })
            .collect();

        let wallet_bytes: Vec<Vec<u8>> = snapshots
            .iter()
            .map(|(w, _, _)| w.as_bytes().to_vec())
            .collect();

        conn.transaction::<_, diesel::result::Error, _>(async |conn| {
            let affected = diesel::insert_into(hp::historical_pnl_snapshots)
                .values(&new_rows)
                .on_conflict((hp::wallet_address, hp::interval_ms, hp::timestamp_ms))
                .do_update()
                .set((
                    hp::total_equity.eq(excluded(hp::total_equity)),
                    hp::attribution.eq(excluded(hp::attribution)),
                ))
                .execute(&mut *conn)
                .await?;

            diesel::sql_query(
                "DELETE FROM historical_pnl_snapshots h
                     USING (
                        SELECT id FROM (
                            SELECT id,
                                   ROW_NUMBER() OVER (
                                       PARTITION BY wallet_address, interval_ms
                                       ORDER BY timestamp_ms DESC
                                   ) AS row_num
                            FROM historical_pnl_snapshots
                            WHERE interval_ms = $1
                              AND wallet_address = ANY($3)
                        ) ranked
                        WHERE ranked.row_num > $2
                     ) stale
                     WHERE h.id = stale.id",
            )
            .bind::<diesel::sql_types::BigInt, _>(interval_ms)
            .bind::<diesel::sql_types::BigInt, _>(max_periods)
            .bind::<diesel::sql_types::Array<diesel::sql_types::Bytea>, _>(&wallet_bytes)
            .execute(&mut *conn)
            .await?;

            Ok(affected)
        })
        .await
        .map_err(Into::into)
    }

    async fn upsert_historical_theo_batch(
        &self,
        interval_ms: i64,
        timestamp_ms: i64,
        snapshots: &[(String, Decimal)],
        max_periods: i64,
    ) -> Result<usize> {
        use schema::historical_theo_snapshots::dsl as ht;

        if snapshots.is_empty() {
            return Ok(0);
        }
        if max_periods <= 0 {
            anyhow::bail!("max_periods must be positive, got {}", max_periods);
        }
        if ![
            HISTORICAL_THEO_INTERVAL_5M_MS,
            HISTORICAL_THEO_INTERVAL_1H_MS,
            HISTORICAL_THEO_INTERVAL_1D_MS,
        ]
        .contains(&interval_ms)
        {
            anyhow::bail!(
                "Invalid interval_ms {}: must be one of the known intervals (5m, 1h, 1d)",
                interval_ms
            );
        }

        let mut conn = self.get_conn().await?;

        let new_rows: Vec<NewHistoricalTheoSnapshot> = snapshots
            .iter()
            .map(|(symbol, theoretical_price)| NewHistoricalTheoSnapshot {
                symbol: symbol.clone(),
                interval_ms,
                timestamp_ms,
                theoretical_price: *theoretical_price,
            })
            .collect();

        let symbols: Vec<String> = snapshots.iter().map(|(symbol, _)| symbol.clone()).collect();

        conn.transaction::<_, diesel::result::Error, _>(async |conn| {
            let affected = diesel::insert_into(ht::historical_theo_snapshots)
                .values(&new_rows)
                .on_conflict((ht::symbol, ht::interval_ms, ht::timestamp_ms))
                .do_update()
                .set(ht::theoretical_price.eq(excluded(ht::theoretical_price)))
                .execute(&mut *conn)
                .await?;

            diesel::sql_query(
                "DELETE FROM historical_theo_snapshots h
                     USING (
                        SELECT id FROM (
                            SELECT id,
                                   ROW_NUMBER() OVER (
                                       PARTITION BY symbol, interval_ms
                                       ORDER BY timestamp_ms DESC
                                   ) AS row_num
                            FROM historical_theo_snapshots
                            WHERE interval_ms = $1
                              AND symbol = ANY($3)
                        ) ranked
                        WHERE ranked.row_num > $2
                     ) stale
                     WHERE h.id = stale.id",
            )
            .bind::<diesel::sql_types::BigInt, _>(interval_ms)
            .bind::<diesel::sql_types::BigInt, _>(max_periods)
            .bind::<diesel::sql_types::Array<diesel::sql_types::Text>, _>(&symbols)
            .execute(&mut *conn)
            .await?;

            Ok(affected)
        })
        .await
        .map_err(Into::into)
    }

    async fn upsert_vol_surface_snapshot(
        &self,
        interval_ms: i64,
        timestamp_ms: i64,
        underlying: &str,
        surface_json: serde_json::Value,
        max_periods: i64,
    ) -> Result<usize> {
        use schema::vol_surface_snapshots::dsl as vs;

        if max_periods <= 0 {
            anyhow::bail!("max_periods must be positive, got {}", max_periods);
        }
        if ![
            HISTORICAL_THEO_INTERVAL_5M_MS,
            HISTORICAL_THEO_INTERVAL_1H_MS,
            HISTORICAL_THEO_INTERVAL_1D_MS,
        ]
        .contains(&interval_ms)
        {
            anyhow::bail!(
                "Invalid interval_ms {}: must be one of the known intervals (5m, 1h, 1d)",
                interval_ms
            );
        }

        let mut conn = self.get_conn().await?;

        let new_row = NewVolSurfaceSnapshotRow {
            underlying: underlying.to_string(),
            interval_ms,
            timestamp_ms,
            surface_json,
        };

        let underlying_owned = underlying.to_string();

        conn.transaction::<_, diesel::result::Error, _>(async |conn| {
            let affected = diesel::insert_into(vs::vol_surface_snapshots)
                .values(&new_row)
                .on_conflict((vs::underlying, vs::interval_ms, vs::timestamp_ms))
                .do_update()
                .set(vs::surface_json.eq(excluded(vs::surface_json)))
                .execute(&mut *conn)
                .await?;

            diesel::sql_query(
                "DELETE FROM vol_surface_snapshots v
                     USING (
                        SELECT id FROM (
                            SELECT id,
                                   ROW_NUMBER() OVER (
                                       PARTITION BY underlying, interval_ms
                                       ORDER BY timestamp_ms DESC
                                   ) AS row_num
                            FROM vol_surface_snapshots
                            WHERE interval_ms = $1
                              AND underlying = $3
                        ) ranked
                        WHERE ranked.row_num > $2
                     ) stale
                     WHERE v.id = stale.id",
            )
            .bind::<diesel::sql_types::BigInt, _>(interval_ms)
            .bind::<diesel::sql_types::BigInt, _>(max_periods)
            .bind::<diesel::sql_types::Text, _>(&underlying_owned)
            .execute(&mut *conn)
            .await?;

            Ok(affected)
        })
        .await
        .map_err(Into::into)
    }

    async fn upsert_bbo_snapshots(
        &self,
        snapshot_ts: i64,
        snapshots: &[NewBboSnapshotInput],
    ) -> Result<usize> {
        use schema::bbo_snapshots::dsl as bbo;

        if snapshots.is_empty() {
            return Ok(0);
        }

        let mut conn = self.get_conn().await?;

        let rows: Vec<NewBboSnapshot> = snapshots
            .iter()
            .map(|snapshot| NewBboSnapshot {
                symbol: snapshot.symbol.clone(),
                best_bid: snapshot.best_bid,
                best_ask: snapshot.best_ask,
                best_bid_size: snapshot.best_bid_size,
                best_ask_size: snapshot.best_ask_size,
                snapshot_ts,
            })
            .collect();

        let affected = diesel::insert_into(bbo::bbo_snapshots)
            .values(&rows)
            .on_conflict((bbo::symbol, bbo::snapshot_ts))
            .do_update()
            .set((
                bbo::best_bid.eq(excluded(bbo::best_bid)),
                bbo::best_ask.eq(excluded(bbo::best_ask)),
                bbo::best_bid_size.eq(excluded(bbo::best_bid_size)),
                bbo::best_ask_size.eq(excluded(bbo::best_ask_size)),
            ))
            .execute(&mut conn)
            .await?;

        Ok(affected)
    }

    async fn delete_bbo_snapshots_older_than(&self, cutoff_ts: i64) -> Result<usize> {
        use schema::bbo_snapshots::dsl as bbo;

        let mut conn = self.get_conn().await?;

        let deleted = diesel::delete(bbo::bbo_snapshots.filter(bbo::snapshot_ts.lt(cutoff_ts)))
            .execute(&mut conn)
            .await?;

        Ok(deleted)
    }

    async fn mark_settlement_payouts_seen(
        &self,
        wallet: &WalletAddress,
        payout_ids: &[i64],
    ) -> Result<usize> {
        use schema::settlement_payout_seen::dsl as sps;
        use schema::settlement_payouts::dsl as sp;

        if payout_ids.is_empty() {
            return Ok(0);
        }

        let mut conn = self.get_conn().await?;

        let valid_ids: Vec<i64> = sp::settlement_payouts
            .filter(sp::wallet.eq(wallet))
            .filter(sp::id.eq_any(payout_ids))
            .select(sp::id)
            .load::<i64>(&mut conn)
            .await?;

        if valid_ids.is_empty() {
            return Ok(0);
        }

        let inserts: Vec<NewSettlementPayoutSeen> = valid_ids
            .into_iter()
            .map(|payout_id| NewSettlementPayoutSeen {
                wallet: *wallet,
                payout_id,
            })
            .collect();

        let inserted_rows = diesel::insert_into(sps::settlement_payout_seen)
            .values(&inserts)
            .on_conflict((sps::wallet, sps::payout_id))
            .do_nothing()
            .execute(&mut conn)
            .await?;

        Ok(inserted_rows)
    }
}

#[cfg(test)]
mod tests {
    use crate::test_helpers::TestDb;
    use hypercall_db::AnalyticsReader;
    use hypercall_db::CatalogReader;
    use hypercall_types::wallet_address::test_wallet;

    #[tokio::test]
    async fn instrument_exists_empty_db_returns_false() {
        let test_db = TestDb::new().await.unwrap();
        let db = test_db.diesel_db().await;
        let exists = db.instrument_exists("BTC-20260131-100000-C").await.unwrap();
        assert!(!exists);
    }

    #[tokio::test]
    async fn get_all_trades_empty_db_returns_empty() {
        let test_db = TestDb::new().await.unwrap();
        let db = test_db.diesel_db().await;
        let trades = db.get_all_trades(100, 0).await.unwrap();
        assert!(trades.is_empty());
    }

    #[tokio::test]
    async fn get_historical_pnl_empty_db_returns_empty() {
        let test_db = TestDb::new().await.unwrap();
        let db = test_db.diesel_db().await;
        let pnl = db
            .get_historical_pnl(
                &test_wallet(60),
                hypercall_types::HISTORICAL_PNL_INTERVAL_1H_MS,
                100,
                false,
            )
            .await
            .unwrap();
        assert!(pnl.is_empty());
    }

    // ---- Convert derive roundtrip tests (no DB needed) ----

    use super::*;
    use hypercall_db::{FillRecord, OrderRecord, TradeRecord};
    use rust_decimal_macros::dec;

    #[test]
    fn trade_row_to_record_roundtrip() {
        let naive_dt = chrono::DateTime::from_timestamp(1_700_000_000, 0)
            .unwrap()
            .naive_utc();
        let row = TradeRow {
            trade_id: 42,
            symbol: "BTC-20260115-95000-C".to_string(),
            price: dec!(100.5),
            size: dec!(1000000),
            maker_address: test_wallet(1),
            taker_address: test_wallet(2),
            maker_fee: dec!(0.5),
            taker_fee: dec!(0.3),
            timestamp: 1_700_000_000_000,
            created_at: Some(naive_dt),
        };
        let record: TradeRecord = row.into();
        assert_eq!(record.trade_id, 42);
        assert_eq!(record.symbol, "BTC-20260115-95000-C");
        assert_eq!(record.price, dec!(100.5));
        assert_eq!(record.size, dec!(1000000));
        assert_eq!(record.maker_address, test_wallet(1));
        assert_eq!(record.taker_address, test_wallet(2));
        assert_eq!(record.maker_fee, dec!(0.5));
        assert_eq!(record.taker_fee, dec!(0.3));
        assert_eq!(record.timestamp, 1_700_000_000_000);
        // NaiveDateTime should be converted to DateTime<Utc>
        let created = record.created_at.expect("created_at should be Some");
        assert_eq!(created.naive_utc(), naive_dt);
    }

    #[test]
    fn trade_row_to_record_none_created_at() {
        let row = TradeRow {
            trade_id: 1,
            symbol: "ETH-20260115-3000-P".to_string(),
            price: dec!(50),
            size: dec!(100),
            maker_address: test_wallet(3),
            taker_address: test_wallet(4),
            maker_fee: dec!(0),
            taker_fee: dec!(0),
            timestamp: 0,
            created_at: None,
        };
        let record: TradeRecord = row.into();
        assert!(record.created_at.is_none());
    }

    #[test]
    fn fill_row_to_record_roundtrip() {
        let naive_dt = chrono::DateTime::from_timestamp(1_700_000_000, 0)
            .unwrap()
            .naive_utc();
        let row = FillRow {
            fill_id: Some(99),
            trade_id: 42,
            wallet_address: test_wallet(5),
            symbol: "BTC-20260115-95000-C".to_string(),
            price: dec!(100.5),
            size: dec!(500),
            fee: dec!(0.25),
            is_taker: true,
            timestamp: 1_700_000_000_000,
            created_at: Some(naive_dt),
            builder_code_address: Some(test_wallet(10)),
            builder_code_fee: Some(dec!(0.1)),
            realized_pnl: Some(dec!(42.5)),
            underlying_notional: Some(dec!(12345)),
        };
        let record: FillRecord = row.into();
        assert_eq!(record.fill_id, Some(99));
        assert_eq!(record.trade_id, 42);
        assert_eq!(record.wallet_address, test_wallet(5));
        assert_eq!(record.symbol, "BTC-20260115-95000-C");
        assert_eq!(record.price, dec!(100.5));
        assert_eq!(record.size, dec!(500));
        assert_eq!(record.fee, dec!(0.25));
        assert!(record.is_taker);
        assert_eq!(record.timestamp, 1_700_000_000_000);
        let created = record.created_at.expect("created_at should be Some");
        assert_eq!(created.naive_utc(), naive_dt);
        assert_eq!(record.builder_code_address, Some(test_wallet(10)));
        assert_eq!(record.builder_code_fee, Some(dec!(0.1)));
        assert_eq!(record.realized_pnl, Some(dec!(42.5)));
        assert_eq!(record.underlying_notional, Some(dec!(12345)));
    }

    #[test]
    fn fill_row_to_record_all_optionals_none() {
        let row = FillRow {
            fill_id: None,
            trade_id: 1,
            wallet_address: test_wallet(6),
            symbol: "ETH-20260115-3000-P".to_string(),
            price: dec!(10),
            size: dec!(100),
            fee: dec!(0),
            is_taker: false,
            timestamp: 0,
            created_at: None,
            builder_code_address: None,
            builder_code_fee: None,
            underlying_notional: None,
            realized_pnl: None,
        };
        let record: FillRecord = row.into();
        assert!(record.fill_id.is_none());
        assert!(record.created_at.is_none());
        assert!(record.builder_code_address.is_none());
        assert!(record.builder_code_fee.is_none());
        assert!(record.realized_pnl.is_none());
        assert!(!record.is_taker);
    }

    #[test]
    fn order_row_to_record_roundtrip() {
        let naive_dt = chrono::DateTime::from_timestamp(1_700_000_000, 0)
            .unwrap()
            .naive_utc();
        let row = OrderRow {
            order_id: 77,
            wallet_address: test_wallet(7),
            symbol: "BTC-20260115-95000-C".to_string(),
            side: "Buy".to_string(),
            price: dec!(200),
            size: dec!(10),
            tif: "GTC".to_string(),
            client_id: Some("my-order-1".to_string()),
            is_perp: false,
            underlying: Some("BTC".to_string()),
            reduce_only: Some(true),
            status: "OPEN".to_string(),
            // In OrderRow: created_at (i64) is renamed to timestamp in OrderRecord
            created_at: 1_700_000_000_000,
            // In OrderRow: updated_at (NaiveDateTime) is renamed to created_at in OrderRecord
            updated_at: naive_dt,
            filled_size: dec!(5),
            mmp_enabled: true,
        };
        let record: OrderRecord = row.into();
        assert_eq!(record.order_id, 77);
        assert_eq!(record.wallet_address, test_wallet(7));
        assert_eq!(record.symbol, "BTC-20260115-95000-C");
        assert_eq!(record.side, "Buy");
        assert_eq!(record.price, dec!(200));
        assert_eq!(record.size, dec!(10));
        assert_eq!(record.tif, "GTC");
        assert_eq!(record.client_id, Some("my-order-1".to_string()));
        assert!(!record.is_perp);
        assert_eq!(record.underlying, Some("BTC".to_string()));
        assert_eq!(record.reduce_only, Some(true));
        assert_eq!(record.status, "OPEN");
        assert_eq!(record.filled_size, dec!(5));
        // Verify rename mapping: row.created_at (i64) -> record.timestamp
        assert_eq!(record.timestamp, 1_700_000_000_000);
        // Verify rename mapping: row.updated_at (NaiveDateTime) -> record.created_at (Option<DateTime<Utc>>)
        let created = record
            .created_at
            .expect("created_at should be Some after conversion");
        assert_eq!(created.naive_utc(), naive_dt);
    }

    #[test]
    fn order_row_to_record_optionals_none() {
        let naive_dt = chrono::DateTime::from_timestamp(1_600_000_000, 0)
            .unwrap()
            .naive_utc();
        let row = OrderRow {
            order_id: 1,
            wallet_address: test_wallet(8),
            symbol: "ETH-20260115-3000-P".to_string(),
            side: "Sell".to_string(),
            price: dec!(50),
            size: dec!(1),
            tif: "IOC".to_string(),
            client_id: None,
            is_perp: true,
            underlying: None,
            reduce_only: None,
            status: "FILLED".to_string(),
            created_at: 1_600_000_000_000,
            updated_at: naive_dt,
            filled_size: dec!(1),
            mmp_enabled: false,
        };
        let record: OrderRecord = row.into();
        assert!(record.client_id.is_none());
        assert!(record.is_perp);
        assert!(record.underlying.is_none());
        assert!(record.reduce_only.is_none());
        assert!(!record.mmp_enabled);
        assert_eq!(record.timestamp, 1_600_000_000_000);
        assert!(record.created_at.is_some());
    }

    // ========================================================================
    // Roundtrip integration tests (insert via raw SQL or Writer, read via trait)
    // ========================================================================

    use hypercall_db::{AnalyticsWriter, NewBboSnapshotInput};
    use hypercall_types::HISTORICAL_PNL_INTERVAL_5M_MS;

    /// Helper: hex-encode a 20-byte wallet address for SQL literals.
    fn wallet_hex(wallet: &hypercall_types::WalletAddress) -> String {
        format!("'\\x{}'", hex::encode(wallet.as_bytes()))
    }

    /// Helper to run raw SQL without ambiguity between sync/async RunQueryDsl.
    fn exec_sql(conn: &mut diesel::pg::PgConnection, sql: &str) {
        diesel::RunQueryDsl::execute(diesel::sql_query(sql), conn).unwrap();
    }

    /// Helper to load rows via raw SQL without ambiguity.
    fn load_sql<T>(conn: &mut diesel::pg::PgConnection, sql: &str) -> Vec<T>
    where
        T: diesel::deserialize::QueryableByName<diesel::pg::Pg> + 'static,
    {
        diesel::RunQueryDsl::load(diesel::sql_query(sql), conn).unwrap()
    }

    #[tokio::test]
    async fn get_all_trades_returns_inserted() {
        let test_db = TestDb::new().await.unwrap();
        let db = test_db.diesel_db().await;

        let mut conn = test_db.handler.pool().get().unwrap();
        let w1 = wallet_hex(&test_wallet(1));
        let w2 = wallet_hex(&test_wallet(2));

        exec_sql(
            &mut conn,
            &format!(
                "INSERT INTO trades (trade_id, symbol, price, size, maker_address, taker_address, maker_fee, taker_fee, timestamp)
                 VALUES
                   (1, 'BTC-20260131-100000-C', 5.0, 1000000, {w1}, {w2}, 0.01, 0.02, 1000),
                   (2, 'BTC-20260131-100000-C', 5.5, 2000000, {w1}, {w2}, 0.01, 0.02, 2000),
                   (3, 'ETH-20260131-4000-P', 1.0, 500000, {w2}, {w1}, 0.005, 0.01, 3000)"
            ),
        );

        let trades = AnalyticsReader::get_all_trades(&db, 10, 0).await.unwrap();
        assert_eq!(trades.len(), 3);
        // Ordered DESC by timestamp
        assert_eq!(trades[0].trade_id, 3);
        assert_eq!(trades[1].trade_id, 2);
        assert_eq!(trades[2].trade_id, 1);
    }

    #[tokio::test]
    async fn get_trades_by_option_filters_correctly() {
        let test_db = TestDb::new().await.unwrap();
        let db = test_db.diesel_db().await;

        let mut conn = test_db.handler.pool().get().unwrap();
        let w1 = wallet_hex(&test_wallet(1));
        let w2 = wallet_hex(&test_wallet(2));

        exec_sql(
            &mut conn,
            &format!(
                "INSERT INTO trades (trade_id, symbol, price, size, maker_address, taker_address, maker_fee, taker_fee, timestamp)
                 VALUES
                   (10, 'BTC-20260131-100000-C', 5.0, 1000000, {w1}, {w2}, 0.01, 0.02, 1000),
                   (11, 'BTC-20260131-100000-C', 6.0, 1000000, {w1}, {w2}, 0.01, 0.02, 2000),
                   (12, 'ETH-20260131-4000-P', 1.0, 500000, {w2}, {w1}, 0.005, 0.01, 3000)"
            ),
        );

        let trades = db
            .get_trades_by_option("BTC-20260131-100000-C", 10)
            .await
            .unwrap();
        assert_eq!(trades.len(), 2);
        // All should be BTC option
        assert!(trades.iter().all(|t| t.symbol == "BTC-20260131-100000-C"));
        // DESC ordering
        assert_eq!(trades[0].trade_id, 11);
        assert_eq!(trades[1].trade_id, 10);

        let eth_trades = db
            .get_trades_by_option("ETH-20260131-4000-P", 10)
            .await
            .unwrap();
        assert_eq!(eth_trades.len(), 1);
        assert_eq!(eth_trades[0].trade_id, 12);
    }

    #[tokio::test]
    async fn get_trades_by_underlying_filters_correctly() {
        let test_db = TestDb::new().await.unwrap();
        let db = test_db.diesel_db().await;

        let mut conn = test_db.handler.pool().get().unwrap();
        let w1 = wallet_hex(&test_wallet(1));
        let w2 = wallet_hex(&test_wallet(2));

        // Insert instruments first (needed for JOIN)
        exec_sql(
            &mut conn,
            "INSERT INTO instruments (id, underlying, strike, expiry, option_type, status)
             VALUES
               ('BTC-20260131-100000-C', 'BTC', 100000, 1738300800, 'C', 'ACTIVE'),
               ('ETH-20260131-4000-P', 'ETH', 4000, 1738300800, 'P', 'ACTIVE')",
        );

        exec_sql(
            &mut conn,
            &format!(
                "INSERT INTO trades (trade_id, symbol, price, size, maker_address, taker_address, maker_fee, taker_fee, timestamp)
                 VALUES
                   (20, 'BTC-20260131-100000-C', 5.0, 1000000, {w1}, {w2}, 0.01, 0.02, 1000),
                   (21, 'BTC-20260131-100000-C', 6.0, 2000000, {w1}, {w2}, 0.01, 0.02, 2000),
                   (22, 'ETH-20260131-4000-P', 1.0, 500000, {w2}, {w1}, 0.005, 0.01, 3000)"
            ),
        );

        let btc_trades = db.get_trades_by_underlying("BTC", 10, 0).await.unwrap();
        assert_eq!(btc_trades.len(), 2);
        assert!(btc_trades
            .iter()
            .all(|t| t.symbol == "BTC-20260131-100000-C"));

        let eth_trades = db.get_trades_by_underlying("ETH", 10, 0).await.unwrap();
        assert_eq!(eth_trades.len(), 1);
        assert_eq!(eth_trades[0].symbol, "ETH-20260131-4000-P");
    }

    #[tokio::test]
    async fn get_fills_by_account_roundtrip() {
        let test_db = TestDb::new().await.unwrap();
        let db = test_db.diesel_db().await;

        let mut conn = test_db.handler.pool().get().unwrap();
        let w1 = wallet_hex(&test_wallet(1));
        let w2 = wallet_hex(&test_wallet(2));
        let wallet1 = test_wallet(1);
        let wallet2 = test_wallet(2);

        // Insert parent trades first (fills reference trade_id FK)
        exec_sql(
            &mut conn,
            &format!(
                "INSERT INTO trades (trade_id, symbol, price, size, maker_address, taker_address, maker_fee, taker_fee, timestamp)
                 VALUES
                   (30, 'BTC-20260131-100000-C', 5.0, 1000000, {w1}, {w2}, 0.01, 0.02, 1000),
                   (31, 'BTC-20260131-100000-C', 6.0, 2000000, {w1}, {w2}, 0.01, 0.02, 2000)"
            ),
        );

        // Insert fills for two wallets
        exec_sql(
            &mut conn,
            &format!(
                "INSERT INTO fills (trade_id, wallet_address, symbol, price, size, fee, is_taker, timestamp)
                 VALUES
                   (30, {w1}, 'BTC-20260131-100000-C', 5.0, 1000000, 0.01, false, 1000),
                   (30, {w2}, 'BTC-20260131-100000-C', 5.0, 1000000, 0.02, true, 1000),
                   (31, {w1}, 'BTC-20260131-100000-C', 6.0, 2000000, 0.01, false, 2000)"
            ),
        );

        let fills_w1 = db.get_fills_by_account(&wallet1, 10, 0).await.unwrap();
        assert_eq!(fills_w1.len(), 2);
        // All for wallet1
        assert!(fills_w1.iter().all(|f| f.wallet_address == wallet1));
        // DESC ordering
        assert_eq!(fills_w1[0].trade_id, 31);
        assert_eq!(fills_w1[1].trade_id, 30);

        let fills_w2 = db.get_fills_by_account(&wallet2, 10, 0).await.unwrap();
        assert_eq!(fills_w2.len(), 1);
        assert_eq!(fills_w2[0].wallet_address, wallet2);
        assert!(fills_w2[0].is_taker);
    }

    #[tokio::test]
    async fn get_orders_by_account_roundtrip() {
        let test_db = TestDb::new().await.unwrap();
        let db = test_db.diesel_db().await;

        let mut conn = test_db.handler.pool().get().unwrap();
        let w1 = wallet_hex(&test_wallet(1));
        let w2 = wallet_hex(&test_wallet(2));
        let wallet1 = test_wallet(1);

        exec_sql(
            &mut conn,
            &format!(
                "INSERT INTO order_infos (order_id, wallet_address, symbol, side, price, size, tif, is_perp, timestamp, status, filled_size, updated_at)
                 VALUES
                   (100, {w1}, 'BTC-20260131-100000-C', 'Buy', 5.0, 1000000, 'GTC', false, 1000, 'OPEN', 0, '2026-01-01 00:00:00'),
                   (101, {w1}, 'BTC-20260131-100000-C', 'Sell', 6.0, 2000000, 'GTC', false, 2000, 'FILLED', 2000000, '2026-01-01 00:01:00'),
                   (102, {w2}, 'ETH-20260131-4000-P', 'Buy', 1.0, 500000, 'IOC', false, 3000, 'CANCELED', 0, '2026-01-01 00:02:00')"
            ),
        );

        let orders = db
            .get_orders_by_account(&wallet1, None, 10, 0)
            .await
            .unwrap();
        assert_eq!(orders.len(), 2);
        assert!(orders.iter().all(|o| o.wallet_address == wallet1));
        // DESC by timestamp
        assert_eq!(orders[0].order_id, 101);
        assert_eq!(orders[1].order_id, 100);
    }

    #[tokio::test]
    async fn get_orders_by_account_with_status_filter() {
        let test_db = TestDb::new().await.unwrap();
        let db = test_db.diesel_db().await;

        let mut conn = test_db.handler.pool().get().unwrap();
        let w1 = wallet_hex(&test_wallet(1));
        let wallet1 = test_wallet(1);

        exec_sql(
            &mut conn,
            &format!(
                "INSERT INTO order_infos (order_id, wallet_address, symbol, side, price, size, tif, is_perp, timestamp, status, filled_size, updated_at)
                 VALUES
                   (200, {w1}, 'BTC-20260131-100000-C', 'Buy', 5.0, 1000000, 'GTC', false, 1000, 'OPEN', 0, '2026-01-01 00:00:00'),
                   (201, {w1}, 'BTC-20260131-100000-C', 'Buy', 5.5, 1000000, 'GTC', false, 2000, 'OPEN', 0, '2026-01-01 00:01:00'),
                   (202, {w1}, 'BTC-20260131-100000-C', 'Sell', 6.0, 2000000, 'GTC', false, 3000, 'FILLED', 2000000, '2026-01-01 00:02:00'),
                   (203, {w1}, 'ETH-20260131-4000-P', 'Buy', 1.0, 500000, 'IOC', false, 4000, 'CANCELED', 0, '2026-01-01 00:03:00')"
            ),
        );

        let open_orders = db
            .get_orders_by_account(&wallet1, Some("OPEN"), 10, 0)
            .await
            .unwrap();
        assert_eq!(open_orders.len(), 2);
        assert!(open_orders.iter().all(|o| o.status == "OPEN"));

        let filled_orders = db
            .get_orders_by_account(&wallet1, Some("FILLED"), 10, 0)
            .await
            .unwrap();
        assert_eq!(filled_orders.len(), 1);
        assert_eq!(filled_orders[0].order_id, 202);

        // Test lowercase normalization
        let canceled = db
            .get_orders_by_account(&wallet1, Some("canceled"), 10, 0)
            .await
            .unwrap();
        assert_eq!(canceled.len(), 1);
        assert_eq!(canceled[0].order_id, 203);
    }

    #[tokio::test]
    async fn upsert_and_get_historical_pnl_roundtrip() {
        let test_db = TestDb::new().await.unwrap();
        let db = test_db.diesel_db().await;

        let wallet = test_wallet(50);
        let interval = HISTORICAL_PNL_INTERVAL_5M_MS;
        let attribution_data = vec![1u8, 2, 3, 4];

        // Call with single wallet, two distinct timestamps
        let snap_t1 = vec![(wallet, dec!(1000.5), Some(attribution_data.clone()))];
        let snap_t2 = vec![(wallet, dec!(1050.25), None)];

        AnalyticsWriter::upsert_historical_pnl_batch(&db, interval, 300_000, &snap_t1, 100)
            .await
            .unwrap();
        AnalyticsWriter::upsert_historical_pnl_batch(&db, interval, 600_000, &snap_t2, 100)
            .await
            .unwrap();

        // Read back (with attribution)
        let points = db
            .get_historical_pnl(&wallet, interval, 100, true)
            .await
            .unwrap();
        assert_eq!(points.len(), 2);
        // Returned in ASC order (reversed from DESC query)
        assert_eq!(points[0].timestamp_ms, 300_000);
        assert_eq!(points[0].total_equity, dec!(1000.5));
        assert_eq!(points[0].attribution, Some(attribution_data));
        assert_eq!(points[1].timestamp_ms, 600_000);
        assert_eq!(points[1].total_equity, dec!(1050.25));
        assert_eq!(points[1].attribution, None);

        // Read without attribution
        let points_no_attr = db
            .get_historical_pnl(&wallet, interval, 100, false)
            .await
            .unwrap();
        assert_eq!(points_no_attr.len(), 2);
        assert!(points_no_attr[0].attribution.is_none());
        assert!(points_no_attr[1].attribution.is_none());
    }

    #[tokio::test]
    async fn upsert_and_get_historical_theos_roundtrip() {
        let test_db = TestDb::new().await.unwrap();
        let db = test_db.diesel_db().await;

        let interval = hypercall_types::HISTORICAL_THEO_INTERVAL_5M_MS;
        let symbol = "BTC-20260131-100000-C".to_string();

        let snap_t1 = vec![(symbol.clone(), dec!(5.25))];
        let snap_t2 = vec![(symbol.clone(), dec!(5.50))];

        AnalyticsWriter::upsert_historical_theo_batch(&db, interval, 300_000, &snap_t1, 100)
            .await
            .unwrap();
        AnalyticsWriter::upsert_historical_theo_batch(&db, interval, 600_000, &snap_t2, 100)
            .await
            .unwrap();

        let points = db
            .get_historical_theos(&symbol, interval, 100)
            .await
            .unwrap();
        assert_eq!(points.len(), 2);
        // Returned in ASC order
        assert_eq!(points[0].timestamp_ms, 300_000);
        assert_eq!(points[0].theoretical_price, dec!(5.25));
        assert_eq!(points[1].timestamp_ms, 600_000);
        assert_eq!(points[1].theoretical_price, dec!(5.50));
    }

    #[tokio::test]
    async fn upsert_and_load_bbo_snapshots_roundtrip() {
        let test_db = TestDb::new().await.unwrap();
        let db = test_db.diesel_db().await;

        let snapshots = vec![
            NewBboSnapshotInput {
                symbol: "BTC-20260131-100000-C".to_string(),
                best_bid: dec!(4.5),
                best_ask: dec!(5.0),
                best_bid_size: Some(dec!(100)),
                best_ask_size: Some(dec!(200)),
                snapshot_ts: 1000,
            },
            NewBboSnapshotInput {
                symbol: "ETH-20260131-4000-P".to_string(),
                best_bid: dec!(0.8),
                best_ask: dec!(1.0),
                best_bid_size: None,
                best_ask_size: None,
                snapshot_ts: 1000,
            },
        ];

        let affected = AnalyticsWriter::upsert_bbo_snapshots(&db, 1000, &snapshots)
            .await
            .unwrap();
        assert_eq!(affected, 2);

        let loaded = db.load_bbo_snapshots_since(0).await.unwrap();
        assert_eq!(loaded.len(), 2);
        // Ordered by symbol ASC, snapshot_ts ASC
        assert_eq!(loaded[0].symbol, "BTC-20260131-100000-C");
        assert_eq!(loaded[0].best_bid, dec!(4.5));
        assert_eq!(loaded[0].best_ask, dec!(5.0));
        assert_eq!(loaded[0].best_bid_size, Some(dec!(100)));
        assert_eq!(loaded[0].best_ask_size, Some(dec!(200)));
        assert_eq!(loaded[0].snapshot_ts, 1000);

        assert_eq!(loaded[1].symbol, "ETH-20260131-4000-P");
        assert_eq!(loaded[1].best_bid, dec!(0.8));
        assert_eq!(loaded[1].best_ask, dec!(1.0));
        assert_eq!(loaded[1].best_bid_size, None);
        assert_eq!(loaded[1].best_ask_size, None);
    }

    #[tokio::test]
    async fn delete_bbo_snapshots_older_than_removes_old() {
        let test_db = TestDb::new().await.unwrap();
        let db = test_db.diesel_db().await;

        let old_snap = vec![NewBboSnapshotInput {
            symbol: "BTC-20260131-100000-C".to_string(),
            best_bid: dec!(4.0),
            best_ask: dec!(4.5),
            best_bid_size: Some(dec!(50)),
            best_ask_size: Some(dec!(60)),
            snapshot_ts: 100,
        }];
        let new_snap = vec![NewBboSnapshotInput {
            symbol: "BTC-20260131-100000-C".to_string(),
            best_bid: dec!(5.0),
            best_ask: dec!(5.5),
            best_bid_size: Some(dec!(70)),
            best_ask_size: Some(dec!(80)),
            snapshot_ts: 2000,
        }];

        AnalyticsWriter::upsert_bbo_snapshots(&db, 100, &old_snap)
            .await
            .unwrap();
        AnalyticsWriter::upsert_bbo_snapshots(&db, 2000, &new_snap)
            .await
            .unwrap();

        // Delete snapshots older than ts=1000
        let deleted = db.delete_bbo_snapshots_older_than(1000).await.unwrap();
        assert_eq!(deleted, 1);

        let remaining = db.load_bbo_snapshots_since(0).await.unwrap();
        assert_eq!(remaining.len(), 1);
        assert_eq!(remaining[0].snapshot_ts, 2000);
    }

    #[tokio::test]
    async fn upsert_and_get_vol_surface_roundtrip() {
        let test_db = TestDb::new().await.unwrap();
        let db = test_db.diesel_db().await;

        let interval = hypercall_types::HISTORICAL_THEO_INTERVAL_5M_MS;
        let surface = serde_json::json!({
            "model": "ssvi",
            "params": [0.1, 0.2, 0.3]
        });

        let affected = AnalyticsWriter::upsert_vol_surface_snapshot(
            &db,
            interval,
            300_000,
            "BTC",
            surface.clone(),
            100,
        )
        .await
        .unwrap();
        assert_eq!(affected, 1);

        // Insert a second snapshot at different timestamp
        let surface2 = serde_json::json!({"model": "sabr", "params": [0.4, 0.5]});
        AnalyticsWriter::upsert_vol_surface_snapshot(
            &db,
            interval,
            600_000,
            "BTC",
            surface2.clone(),
            100,
        )
        .await
        .unwrap();

        let history = db
            .get_vol_surface_history("BTC", interval, 100)
            .await
            .unwrap();
        assert_eq!(history.len(), 2);
        // Returned in ASC order
        assert_eq!(history[0].timestamp_ms, 300_000);
        assert_eq!(history[0].surface_json, surface);
        assert_eq!(history[0].underlying, "BTC");
        assert_eq!(history[1].timestamp_ms, 600_000);
        assert_eq!(history[1].surface_json, surface2);
    }

    #[tokio::test]
    async fn mark_and_get_seen_settlement_payouts() {
        let test_db = TestDb::new().await.unwrap();
        let db = test_db.diesel_db().await;

        let mut conn = test_db.handler.pool().get().unwrap();
        let wallet = test_wallet(70);
        let w_hex = wallet_hex(&wallet);

        // Insert settlement payouts directly
        exec_sql(
            &mut conn,
            &format!(
                "INSERT INTO settlement_payouts (wallet, symbol, expiry_ts, position_size, settlement_price, payout_amount, ledger_applied)
                 VALUES
                   ({w_hex}, 'BTC-20260131-100000-C', 1738300800, 1000000, 100000, 5000, true),
                   ({w_hex}, 'ETH-20260131-4000-P', 1738300800, 500000, 4000, 200, true)"
            ),
        );

        // Get the IDs of inserted payouts
        #[derive(diesel::QueryableByName)]
        struct IdRow {
            #[diesel(sql_type = diesel::sql_types::BigInt)]
            id: i64,
        }
        let ids: Vec<IdRow> = load_sql(
            &mut conn,
            &format!("SELECT id FROM settlement_payouts WHERE wallet = {w_hex} ORDER BY id"),
        );
        let payout_ids: Vec<i64> = ids.iter().map(|r| r.id).collect();
        assert_eq!(payout_ids.len(), 2);

        // Before marking, none should be seen
        let seen_before = db
            .get_seen_settlement_payout_ids(&wallet, &payout_ids)
            .await
            .unwrap();
        assert!(seen_before.is_empty());

        // Mark first payout as seen
        let marked = AnalyticsWriter::mark_settlement_payouts_seen(&db, &wallet, &[payout_ids[0]])
            .await
            .unwrap();
        assert_eq!(marked, 1);

        // Verify only first is seen
        let seen_after = db
            .get_seen_settlement_payout_ids(&wallet, &payout_ids)
            .await
            .unwrap();
        assert_eq!(seen_after.len(), 1);
        assert!(seen_after.contains(&payout_ids[0]));
        assert!(!seen_after.contains(&payout_ids[1]));

        // Mark both (idempotent for already-seen)
        AnalyticsWriter::mark_settlement_payouts_seen(&db, &wallet, &payout_ids)
            .await
            .unwrap();
        let seen_all = db
            .get_seen_settlement_payout_ids(&wallet, &payout_ids)
            .await
            .unwrap();
        assert_eq!(seen_all.len(), 2);
    }

    #[tokio::test]
    async fn instrument_exists_true_after_insert() {
        let test_db = TestDb::new().await.unwrap();
        let db = test_db.diesel_db().await;

        let mut conn = test_db.handler.pool().get().unwrap();

        // Before insert
        assert!(!db.instrument_exists("BTC-20260131-100000-C").await.unwrap());

        exec_sql(
            &mut conn,
            "INSERT INTO instruments (id, underlying, strike, expiry, option_type, status)
             VALUES ('BTC-20260131-100000-C', 'BTC', 100000, 1738300800, 'C', 'ACTIVE')",
        );

        // After insert
        assert!(db.instrument_exists("BTC-20260131-100000-C").await.unwrap());
        // Different symbol still false
        assert!(!db.instrument_exists("ETH-20260131-4000-P").await.unwrap());
    }

    #[tokio::test]
    async fn get_client_ids_by_order_ids_roundtrip() {
        let test_db = TestDb::new().await.unwrap();
        let db = test_db.diesel_db().await;

        let mut conn = test_db.handler.pool().get().unwrap();
        let w1 = wallet_hex(&test_wallet(1));

        exec_sql(
            &mut conn,
            &format!(
                "INSERT INTO order_infos (order_id, wallet_address, symbol, side, price, size, tif, is_perp, timestamp, status, filled_size, updated_at, client_id)
                 VALUES
                   (300, {w1}, 'BTC-20260131-100000-C', 'Buy', 5.0, 1000000, 'GTC', false, 1000, 'OPEN', 0, '2026-01-01 00:00:00', 'client-alpha'),
                   (301, {w1}, 'BTC-20260131-100000-C', 'Sell', 6.0, 2000000, 'GTC', false, 2000, 'OPEN', 0, '2026-01-01 00:01:00', NULL),
                   (302, {w1}, 'ETH-20260131-4000-P', 'Buy', 1.0, 500000, 'IOC', false, 3000, 'FILLED', 500000, '2026-01-01 00:02:00', 'client-beta')"
            ),
        );

        let result = db
            .get_client_ids_by_order_ids(&[300, 301, 302, 999])
            .await
            .unwrap();
        assert_eq!(result.len(), 3); // 999 not found
        assert_eq!(result[&300], Some("client-alpha".to_string()));
        assert_eq!(result[&301], None);
        assert_eq!(result[&302], Some("client-beta".to_string()));
        assert!(!result.contains_key(&999));
    }

    // ========================================================================
    // Tests for newly added trait methods
    // ========================================================================

    #[tokio::test]
    async fn get_theo_marks_at_timestamp_picks_latest_before() {
        let test_db = TestDb::new().await.unwrap();
        let db = test_db.diesel_db().await;

        let interval = hypercall_types::HISTORICAL_THEO_INTERVAL_5M_MS;
        let symbol = "BTC-20260131-100000-C".to_string();

        // Insert theos at different timestamps
        AnalyticsWriter::upsert_historical_theo_batch(
            &db,
            interval,
            1000,
            &[(symbol.clone(), dec!(5.0))],
            100,
        )
        .await
        .unwrap();

        AnalyticsWriter::upsert_historical_theo_batch(
            &db,
            interval,
            2000,
            &[(symbol.clone(), dec!(7.5))],
            100,
        )
        .await
        .unwrap();

        AnalyticsWriter::upsert_historical_theo_batch(
            &db,
            interval,
            3000,
            &[(symbol.clone(), dec!(9.0))],
            100,
        )
        .await
        .unwrap();

        // Query at ts=2500 should pick ts=2000 (latest before 2500)
        let marks = db
            .get_theo_marks_at_timestamp(&[symbol.clone()], 2500)
            .await
            .unwrap();
        assert_eq!(marks.len(), 1);
        assert_eq!(marks[&symbol], dec!(7.5));

        // Query at ts=3000 should pick ts=3000 (equal counts)
        let marks2 = db
            .get_theo_marks_at_timestamp(&[symbol.clone()], 3000)
            .await
            .unwrap();
        assert_eq!(marks2[&symbol], dec!(9.0));

        // Query at ts=500 should return empty (no data before 500)
        let marks3 = db
            .get_theo_marks_at_timestamp(&[symbol.clone()], 500)
            .await
            .unwrap();
        assert!(marks3.is_empty());
    }

    #[tokio::test]
    async fn get_theo_marks_at_timestamp_empty_symbols() {
        let test_db = TestDb::new().await.unwrap();
        let db = test_db.diesel_db().await;

        let marks = db.get_theo_marks_at_timestamp(&[], 1000).await.unwrap();
        assert!(marks.is_empty());
    }

    #[tokio::test]
    async fn get_deposit_withdraw_events_returns_ordered() {
        let test_db = TestDb::new().await.unwrap();
        let db = test_db.diesel_db().await;

        let mut conn = test_db.handler.pool().get().unwrap();
        let wallet = test_wallet(80);
        let w_hex = wallet_hex(&wallet);

        // Insert deposit and withdraw events
        exec_sql(
            &mut conn,
            &format!(
                "INSERT INTO ledger_events (wallet, event_ts_ms, delta, event_type) VALUES \
                 ({w_hex}, 1000, 500.0, 'deposit'), \
                 ({w_hex}, 2000, -100.0, 'withdraw'), \
                 ({w_hex}, 3000, 200.0, 'deposit'), \
                 ({w_hex}, 2500, 50.0, 'fill_premium')"
            ),
        );

        let events = db.get_deposit_withdraw_events(&wallet).await.unwrap();
        // Should only return deposit and withdraw, not fill_premium
        assert_eq!(events.len(), 3);
        // Ordered by event_ts_ms ASC
        assert_eq!(events[0].event_ts_ms, 1000);
        assert_eq!(events[0].delta, dec!(500.0));
        assert_eq!(events[1].event_ts_ms, 2000);
        assert_eq!(events[1].delta, dec!(-100.0));
        assert_eq!(events[2].event_ts_ms, 3000);
        assert_eq!(events[2].delta, dec!(200.0));
    }

    #[tokio::test]
    async fn get_deposit_withdraw_events_empty() {
        let test_db = TestDb::new().await.unwrap();
        let db = test_db.diesel_db().await;

        let wallet = test_wallet(81);
        let events = db.get_deposit_withdraw_events(&wallet).await.unwrap();
        assert!(events.is_empty());
    }

    #[tokio::test]
    async fn get_settled_pnl_by_symbol_groups_correctly() {
        let test_db = TestDb::new().await.unwrap();
        let db = test_db.diesel_db().await;

        let mut conn = test_db.handler.pool().get().unwrap();
        let wallet = test_wallet(82);
        let w_hex = wallet_hex(&wallet);

        // Insert ledger events with different event types and reference symbols
        exec_sql(
            &mut conn,
            &format!(
                "INSERT INTO ledger_events (wallet, event_ts_ms, delta, event_type, reference_symbol) VALUES \
                 ({w_hex}, 1000, 100.0, 'fill_premium', 'BTC-20260131-100000-C'), \
                 ({w_hex}, 1500, -30.0, 'fill_premium', 'BTC-20260131-100000-C'), \
                 ({w_hex}, 2000, 50.0, 'settlement_realized_pnl', 'ETH-20260131-4000-P'), \
                 ({w_hex}, 2500, 25.0, 'fill_realized_pnl', 'ETH-20260131-4000-P'), \
                 ({w_hex}, 3000, 200.0, 'deposit', NULL)"
            ),
        );

        // Query before ts=3000 should include all matching events
        let pnl = db.get_settled_pnl_by_symbol(&wallet, 3000).await.unwrap();
        assert_eq!(pnl.len(), 2);

        let btc_pnl = pnl.iter().find(|(s, _)| s == "BTC-20260131-100000-C");
        assert_eq!(btc_pnl.unwrap().1, dec!(70.0)); // 100 - 30

        let eth_pnl = pnl.iter().find(|(s, _)| s == "ETH-20260131-4000-P");
        assert_eq!(eth_pnl.unwrap().1, dec!(75.0)); // 50 + 25
    }

    #[tokio::test]
    async fn get_settled_pnl_by_symbol_respects_timestamp_cutoff() {
        let test_db = TestDb::new().await.unwrap();
        let db = test_db.diesel_db().await;

        let mut conn = test_db.handler.pool().get().unwrap();
        let wallet = test_wallet(83);
        let w_hex = wallet_hex(&wallet);

        exec_sql(
            &mut conn,
            &format!(
                "INSERT INTO ledger_events (wallet, event_ts_ms, delta, event_type, reference_symbol) VALUES \
                 ({w_hex}, 1000, 100.0, 'fill_premium', 'BTC-20260131-100000-C'), \
                 ({w_hex}, 5000, 200.0, 'fill_premium', 'BTC-20260131-100000-C')"
            ),
        );

        // Query before ts=2000 should only include the first event
        let pnl = db.get_settled_pnl_by_symbol(&wallet, 2000).await.unwrap();
        assert_eq!(pnl.len(), 1);
        assert_eq!(pnl[0].1, dec!(100.0));
    }

    #[tokio::test]
    async fn get_fills_since_timestamp_filters_correctly() {
        let test_db = TestDb::new().await.unwrap();
        let db = test_db.diesel_db().await;

        let mut conn = test_db.handler.pool().get().unwrap();
        let w1 = wallet_hex(&test_wallet(1));
        let w2 = wallet_hex(&test_wallet(2));

        // Insert parent trades first
        exec_sql(
            &mut conn,
            &format!(
                "INSERT INTO trades (trade_id, symbol, price, size, maker_address, taker_address, maker_fee, taker_fee, timestamp) \
                 VALUES \
                   (40, 'BTC-20260131-100000-C', 5.0, 1000000, {w1}, {w2}, 0.01, 0.02, 1000), \
                   (41, 'BTC-20260131-100000-C', 6.0, 2000000, {w1}, {w2}, 0.01, 0.02, 2000), \
                   (42, 'ETH-20260131-4000-P', 1.0, 500000, {w2}, {w1}, 0.005, 0.01, 3000)"
            ),
        );

        // Insert fills at different timestamps. Market-stats backfill reads
        // only taker rows so it does not double-count both wallet fill rows.
        exec_sql(
            &mut conn,
            &format!(
                "INSERT INTO fills (trade_id, wallet_address, symbol, price, size, fee, is_taker, timestamp, underlying_notional) \
                 VALUES \
                   (40, {w1}, 'BTC-20260131-100000-C', 5.0, 1000000, 0.01, false, 1000, 100.0), \
                   (41, {w1}, 'BTC-20260131-100000-C', 6.0, 2000000, 0.01, false, 2000, 200.0), \
                   (42, {w2}, 'ETH-20260131-4000-P', 1.0, 500000, 0.02, true, 3000, 300.0)"
            ),
        );

        // Query since ts=2000 should return only the taker fill at 3000.
        let fills = db.get_fills_since_timestamp(2000).await.unwrap();
        assert_eq!(fills.len(), 1);
        // Ordered ASC by timestamp
        assert_eq!(fills[0].timestamp, 3000);
        assert_eq!(fills[0].symbol, "ETH-20260131-4000-P");
        assert_eq!(fills[0].underlying_notional, Some(dec!(300.0)));

        // Query since ts=4000 should return empty
        let fills_empty = db.get_fills_since_timestamp(4000).await.unwrap();
        assert!(fills_empty.is_empty());
    }
}