sqlite.rs

use anyhow::{Context, Result};
use rust_decimal::Decimal;
use rusqlite::{Connection, params};
use std::str::FromStr;
use uuid::Uuid;

use crate::types::PositionState;
pub use super::types::*;

/// Normalize Decimal to string: strip trailing zeros so all exchanges produce
/// consistent representations (e.g. "18.0" → "18", "0.5300" → "0.53").
#[inline]
fn dec(d: Decimal) -> String {
    d.normalize().to_string()
}

/// Derived PnL breakdown for a closed trade — computed from per-order fill
/// data + trade-level funding. Single source of truth for any UI/stats that
/// show realized PnL.
///
/// Returns `(gross_pnl, net_pnl, net_pnl_pct)` where:
///   gross = (exit_1 - entry_1) × qty + (entry_2 - exit_2) × qty  (is_long_1)
///   net   = gross - fees + funding_total
///   pct   = net / avg_notional × 100
///
/// For partial/failed closes (any price zero), gross and pct are zero —
/// matching the old executor's "unhedged close" fallback.
fn derive_pnl(
    is_long_1: bool,
    qty: &str,
    entry_1: &str, entry_2: &str,
    exit_1: &str, exit_2: &str,
    fees: &str,
    funding_1: &str, funding_2: &str,
) -> (Decimal, Decimal, Decimal) {
    use rust_decimal_macros::dec as decm;
    // Trim before parse — SQLite's CAST ... AS REAL tolerates leading
    // whitespace, so callers that previously went through raw SQL might
    // hand us " 99" etc. Strict `Decimal::from_str` would silently
    // collapse those to zero and trigger the `both_filled` fallback.
    let q = Decimal::from_str(qty.trim()).unwrap_or(Decimal::ZERO);
    let e1 = Decimal::from_str(entry_1.trim()).unwrap_or(Decimal::ZERO);
    let e2 = Decimal::from_str(entry_2.trim()).unwrap_or(Decimal::ZERO);
    let x1 = Decimal::from_str(exit_1.trim()).unwrap_or(Decimal::ZERO);
    let x2 = Decimal::from_str(exit_2.trim()).unwrap_or(Decimal::ZERO);
    let f  = Decimal::from_str(fees.trim()).unwrap_or(Decimal::ZERO);
    let fd1 = Decimal::from_str(funding_1.trim()).unwrap_or(Decimal::ZERO);
    let fd2 = Decimal::from_str(funding_2.trim()).unwrap_or(Decimal::ZERO);

    // Partial/failed close: any leg missing a price → gross unknown.
    let both_filled = !e1.is_zero() && !e2.is_zero() && !x1.is_zero() && !x2.is_zero();
    let gross = if !both_filled {
        Decimal::ZERO
    } else if is_long_1 {
        (x1 - e1) * q + (e2 - x2) * q
    } else {
        (e1 - x1) * q + (x2 - e2) * q
    };

    let net = gross - f + fd1 + fd2;
    let notional = if both_filled { q * (e1 + e2) / decm!(2) } else { Decimal::ZERO };
    let net_pct = if notional.is_zero() { Decimal::ZERO } else { net / notional * decm!(100) };
    (gross, net, net_pct)
}

// ── Storage ──────────────────────────────────────────────────────────────────

/// One leg of a reconciliation close — the fill data needed to persist an
/// `orders` row alongside the trade state transition. Passed to
/// `finalize_reconciliation_close` so both writes happen in one transaction.
pub struct ReconcileLegFill {
    pub order_id: u64,
    pub exchange_id: i64,
    pub purpose: String,
    pub side: String,
    pub qty: Decimal,
    pub fill_price: Decimal,
    pub fee: Decimal,
}

pub struct Storage {
    conn: Connection,
}

impl Storage {
    /// Open a read-only connection (no schema init, no WAL setup).
    /// Safe for concurrent reads alongside the main write connection.
    pub fn open_read_only(path: &str) -> Result<Self> {
        let conn = Connection::open_with_flags(
            path,
            rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY | rusqlite::OpenFlags::SQLITE_OPEN_NO_MUTEX,
        ).with_context(|| format!("Failed to open read-only SQLite at {path}"))?;
        conn.pragma_update(None, "busy_timeout", 5000)?;
        Ok(Self { conn })
    }

    pub fn open(path: &str) -> Result<Self> {
        if let Some(parent) = std::path::Path::new(path).parent() {
            std::fs::create_dir_all(parent)?;
        }

        let conn = Connection::open(path)
            .with_context(|| format!("Failed to open SQLite database at {path}"))?;

        conn.pragma_update(None, "journal_mode", "WAL")?;
        conn.pragma_update(None, "synchronous", "NORMAL")?;
        conn.pragma_update(None, "busy_timeout", 5000)?;
        conn.pragma_update(None, "foreign_keys", "ON")?;

        conn.execute_batch(SCHEMA)?;

        // Seed known exchanges with stable IDs.
        let now_ms = crate::utils::time::current_unix_ms();
        conn.execute(
            "INSERT OR IGNORE INTO exchanges (id, name, display_name, exchange_type, added_at_ms)
             VALUES (1, 'lighter', 'Lighter.xyz', 'dex_onchain', ?1)",
            params![now_ms],
        )?;
        conn.execute(
            "INSERT OR IGNORE INTO exchanges (id, name, display_name, exchange_type, added_at_ms)
             VALUES (2, 'extended', 'Extended.exchange', 'dex_offchain', ?1)",
            params![now_ms],
        )?;
        // zero1 = 01.xyz / N1 / Nord engine — Solana-based perpetual DEX. Display
        // label "01" matches the brand on the website; module name "zero1" is
        // used everywhere in code/config/secrets to keep grep clean and avoid
        // TOML quoting (a section name starting with a digit must be quoted).
        conn.execute(
            "INSERT OR IGNORE INTO exchanges (id, name, display_name, exchange_type, added_at_ms)
             VALUES (3, 'zero1', '01', 'dex_offchain', ?1)",
            params![now_ms],
        )?;
        conn.execute(
            "INSERT OR IGNORE INTO exchanges (id, name, display_name, exchange_type, added_at_ms)
             VALUES (4, 'hyperliquid', 'Hyperliquid', 'dex_onchain', ?1)",
            params![now_ms],
        )?;
        conn.execute(
            "INSERT OR IGNORE INTO exchanges (id, name, display_name, exchange_type, added_at_ms)
             VALUES (5, 'bybit', 'Bybit', 'cex', ?1)",
            params![now_ms],
        )?;
        // Additional exchanges are registered via ensure_exchange() at startup.

        // Migrations (idempotent ALTER TABLE pattern):
        let _ = conn.execute_batch("ALTER TABLE orders ADD COLUMN signal_price TEXT DEFAULT '0';");
        let _ = conn.execute_batch("ALTER TABLE exchange_symbols ADD COLUMN limit_price_cap REAL;");
        let _ = conn.execute_batch("ALTER TABLE trades ADD COLUMN entry_spread_bps TEXT DEFAULT '0';");
        // Strategy config snapshot columns (replaces config_json blob).
        let _ = conn.execute_batch("ALTER TABLE trades ADD COLUMN cfg_entry_bps TEXT;");
        let _ = conn.execute_batch("ALTER TABLE trades ADD COLUMN cfg_include_fees_in_entry INTEGER;");
        let _ = conn.execute_batch("ALTER TABLE trades ADD COLUMN cfg_exit_mode TEXT;");
        let _ = conn.execute_batch("ALTER TABLE trades ADD COLUMN cfg_exit_bps TEXT;");
        let _ = conn.execute_batch("ALTER TABLE trades ADD COLUMN cfg_exit_pnl_pct TEXT;");
        let _ = conn.execute_batch("ALTER TABLE trades ADD COLUMN cfg_include_fees_in_exit INTEGER;");
        let _ = conn.execute_batch("ALTER TABLE trades ADD COLUMN cfg_leverage INTEGER;");
        // Drop legacy config_json column (replaced by cfg_* typed columns).
        // SQLite 3.35+ (2021) supports DROP COLUMN; older versions will silently no-op.
        let _ = conn.execute_batch("ALTER TABLE trades DROP COLUMN config_json;");
        let _ = conn.execute_batch("ALTER TABLE optimizer_results ADD COLUMN min_exit_spread_bps REAL;");
        let _ = conn.execute_batch("ALTER TABLE optimizer_results ADD COLUMN entry_spike_bps REAL;");
        let _ = conn.execute_batch("ALTER TABLE optimizer_results ADD COLUMN exit_spike_bps REAL;");
        let _ = conn.execute_batch("ALTER TABLE optimizer_results ADD COLUMN max_pos_per_symbol INTEGER;");
        // Per-position funding tracking + SL trigger price persistence.
        let _ = conn.execute_batch("ALTER TABLE trades ADD COLUMN funding_1 TEXT DEFAULT '0';");
        let _ = conn.execute_batch("ALTER TABLE trades ADD COLUMN funding_2 TEXT DEFAULT '0';");
        let _ = conn.execute_batch("ALTER TABLE orders ADD COLUMN trigger_price TEXT DEFAULT '0';");
        let _ = conn.execute_batch("ALTER TABLE orders ADD COLUMN updated_at_ms INTEGER NOT NULL DEFAULT 0;");
        let _ = conn.execute_batch("ALTER TABLE orders ADD COLUMN move_count INTEGER NOT NULL DEFAULT 0;");
        // 2026-04-14: Per-leg local wall-clock recv timestamp at signal time
        // (ms). Mirrors `exchange_ts_ms` but measured in our own clock so the
        // `Recv→Sig` UI metric has uniform semantics across all venues,
        // including ZERO1 which does not report event time.
        let _ = conn.execute_batch("ALTER TABLE orders ADD COLUMN recv_ts_ms INTEGER DEFAULT 0;");
        // 2026-04-14: Time (ms, local wall clock) when the close signal fired
        // for this trade. Symmetric to entry-side `signal_at_ms`. Used by the
        // Recv→Sig UI metric for close orders — without it, close orders
        // incorrectly reused the entry signal time (hours stale for any
        // long-held position).
        let _ = conn.execute_batch("ALTER TABLE trades ADD COLUMN exit_signal_at_ms INTEGER DEFAULT 0;");
        // Per-pair symbol blocking: second exchange ID for pair-specific blocks.
        let _ = conn.execute_batch("ALTER TABLE symbol_blocks ADD COLUMN exchange_id_2 INTEGER REFERENCES exchanges(id);");
        // Wall-clock receive-time staleness filter (added 2026-04-10 after the
        // LIT/AVAX phantom-signal incident — see commit 1f6bad0 + follow-up).
        let _ = conn.execute_batch("ALTER TABLE symbol_overrides ADD COLUMN max_recv_staleness_secs REAL;");
        // 2026-04-14: replace `max_recv_age_secs` (Recv→Sig only) with
        // `max_tick_age_ms` which covers Tick→Recv + Recv→Sig in a single
        // gate via `last_exchange_ts`. No backward compat — old values drop.
        let _ = conn.execute_batch("ALTER TABLE symbol_overrides DROP COLUMN max_recv_age_secs;");
        let _ = conn.execute_batch("ALTER TABLE symbol_overrides ADD COLUMN max_tick_age_ms REAL;");
        // 2026-04-10: drop legacy freshness/gap filter columns. The bot's
        // runtime spread_detector and the sim both use the recv-time filters
        // exclusively now; these columns are dead. SQLite ≥3.35 supports
        // DROP COLUMN; older versions silently no-op (which is fine — the
        // columns become unused but harmless).
        let _ = conn.execute_batch("ALTER TABLE symbol_overrides DROP COLUMN max_tick_staleness_secs;");
        let _ = conn.execute_batch("ALTER TABLE symbol_overrides DROP COLUMN max_tick_age_secs;");
        let _ = conn.execute_batch("ALTER TABLE symbol_overrides DROP COLUMN entry_gap_secs;");
        let _ = conn.execute_batch("ALTER TABLE symbol_overrides DROP COLUMN exit_gap_secs;");
        // 2026-04-13: per-exchange halt scope. NULL = global halt
        // (KillSwitch/ManualPause/ApiErrorsPermanent), CSV of i64 db_ids =
        // halt only blocks signals touching those exchanges.
        let _ = conn.execute_batch("ALTER TABLE trading_halts ADD COLUMN affected_exchanges TEXT;");
        // 2026-04-15: drop trades.fees column. It was a synthetic value computed
        // from config fee_rate × notional and diverged from the actual exchange
        // fees stored per-order in orders.fee. Single source of truth now:
        // SUM(orders.fee) for every read. See load_* functions for the subquery.
        let _ = conn.execute_batch("ALTER TABLE trades DROP COLUMN fees;");
        // 2026-04-15: drop trades.net_pnl and trades.net_pnl_pct. Same reason as
        // `fees` above — both are pure derivatives of (orders entry prices,
        // orders exit prices, orders.fee, trades.qty, trades.is_long_1,
        // trades.funding_1/2). Anything trade-level is already stored (funding,
        // qty, direction); PnL is just arithmetic over orders + those fields.
        // Keeping it as a column meant a parallel source of truth that
        // inevitably drifted. Reads now derive net PnL inline from the same
        // JOINs that already fetch entry/exit prices.
        let _ = conn.execute_batch("ALTER TABLE trades DROP COLUMN net_pnl;");
        let _ = conn.execute_batch("ALTER TABLE trades DROP COLUMN net_pnl_pct;");

        // Drop legacy CSV-era columns from exchange_symbols (market_params.csv removed).
        let _ = conn.execute_batch("ALTER TABLE exchange_symbols DROP COLUMN quote_asset;");
        let _ = conn.execute_batch("ALTER TABLE exchange_symbols DROP COLUMN price_step;");
        let _ = conn.execute_batch("ALTER TABLE exchange_symbols DROP COLUMN mmf_updated_at;");
        let _ = conn.execute_batch("ALTER TABLE exchange_symbols DROP COLUMN status;");

        // 2026-04-17: merge entry_tick_max_spike_bps + exit_tick_max_spike_bps
        // into a single tick_max_spike_bps. Both servers have 0 rows so no
        // data migration needed — just reshape the schema.
        let _ = conn.execute_batch("ALTER TABLE symbol_overrides DROP COLUMN entry_tick_max_spike_bps;");
        let _ = conn.execute_batch("ALTER TABLE symbol_overrides DROP COLUMN exit_tick_max_spike_bps;");
        let _ = conn.execute_batch("ALTER TABLE symbol_overrides ADD COLUMN tick_max_spike_bps REAL;");

        // Mark stale optimizer runs as interrupted (from previous crash/restart)
        let _ = conn.execute(
            "UPDATE optimizer_runs SET status = 'interrupted' WHERE status = 'running'",
            [],
        );

        Ok(Self { conn })
    }

    /// Direct access to the connection (for early startup queries before full API).
    pub fn conn_ref(&self) -> &Connection {
        &self.conn
    }

    // ── Symbols ──────────────────────────────────────────────────────────────

    /// Upsert one row per (symbol, exchange_id) into exchange_symbols.
    /// Data comes from exchange REST APIs at startup (via `MarketMeta`).
    pub fn upsert_exchange_symbol(&self, s: &ExchangeSymbolConfig) -> Result<()> {
        self.conn.execute(
            "INSERT INTO exchange_symbols (
                symbol, exchange_id, market_name,
                size_decimals, price_decimals,
                qty_step, min_qty, max_leverage,
                mmf, limit_price_cap
            ) VALUES (?1,?2,?3,?4,?5,?6,?7,?8,?9,?10)
            ON CONFLICT(symbol, exchange_id) DO UPDATE SET
                market_name=COALESCE(NULLIF(excluded.market_name,''), exchange_symbols.market_name),
                size_decimals=excluded.size_decimals,
                price_decimals=excluded.price_decimals,
                qty_step=excluded.qty_step,
                min_qty=excluded.min_qty,
                max_leverage=excluded.max_leverage,
                mmf=COALESCE(excluded.mmf, exchange_symbols.mmf),
                limit_price_cap=COALESCE(excluded.limit_price_cap, exchange_symbols.limit_price_cap)",
            params![
                s.symbol, s.exchange_id, s.market_name,
                s.size_decimals, s.price_decimals,
                s.qty_step, s.min_qty, s.max_leverage,
                s.mmf, s.limit_price_cap,
            ],
        )?;
        Ok(())
    }

    /// Load all exchange_symbols rows.
    pub fn load_exchange_symbols(&self) -> Result<Vec<ExchangeSymbolConfig>> {
        let mut stmt = self.conn.prepare(
            "SELECT symbol, exchange_id, market_name,
                    size_decimals, price_decimals,
                    qty_step, min_qty, max_leverage,
                    mmf, limit_price_cap
             FROM exchange_symbols"
        )?;
        let rows = stmt.query_map([], |row| {
            Ok(ExchangeSymbolConfig {
                symbol: row.get(0)?,
                exchange_id: row.get(1)?,
                market_name: row.get::<_, String>(2).unwrap_or_default(),
                size_decimals: row.get(3)?,
                price_decimals: row.get(4)?,
                qty_step: row.get(5)?,
                min_qty: row.get(6)?,
                max_leverage: row.get(7)?,
                mmf: row.get(8)?,
                limit_price_cap: row.get(9)?,
            })
        })?;
        rows.collect::<std::result::Result<Vec<_>, _>>()
            .context("Failed to load exchange symbols")
    }

    /// Returns (exchange_id, max_leverage) pairs for a given symbol.
    pub fn get_max_leverage_for_symbol(&self, symbol: &str) -> Result<Vec<(i64, i32)>> {
        let mut stmt = self.conn.prepare(
            "SELECT exchange_id, max_leverage FROM exchange_symbols WHERE symbol = ?1"
        )?;
        let rows = stmt.query_map([symbol], |row| {
            Ok((row.get::<_, i64>(0)?, row.get::<_, i32>(1)?))
        })?;
        rows.collect::<std::result::Result<Vec<_>, _>>()
            .context("Failed to load max leverage for symbol")
    }

    // ── Operator overrides ────────────────────────────────────────────────────

    /// Upsert an operator override for (symbol, exchange_id, field).
    /// Used by web UI / CLI to set manual mmf or api_alias overrides.
    pub fn upsert_override(&self, symbol: &str, exchange_id: i64, field: &str, value: &str) -> Result<()> {
        self.conn.execute(
            "INSERT INTO exchange_symbol_overrides (symbol, exchange_id, field_name, value)
             VALUES (?1, ?2, ?3, ?4)
             ON CONFLICT(symbol, exchange_id, field_name) DO UPDATE SET
                value=excluded.value, updated_at=datetime('now')",
            params![symbol, exchange_id, field, value],
        )?;
        Ok(())
    }

    /// Load all overrides as (symbol, exchange_id, field, value) tuples.
    pub fn load_overrides(&self) -> Result<Vec<(String, i64, String, String)>> {
        let mut stmt = self.conn.prepare(
            "SELECT symbol, exchange_id, field_name, value FROM exchange_symbol_overrides"
        )?;
        let rows = stmt.query_map([], |row| {
            Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?))
        })?;
        rows.collect::<std::result::Result<Vec<_>, _>>()
            .context("Failed to load overrides")
    }

    /// Load api_alias overrides → HashMap<alias, canonical_symbol>.
    pub fn load_api_aliases(&self, exchange_id: i64) -> Result<std::collections::HashMap<String, String>> {
        let mut stmt = self.conn.prepare(
            "SELECT value, symbol FROM exchange_symbol_overrides
             WHERE exchange_id = ?1 AND field_name = 'api_alias'"
        )?;
        let rows = stmt.query_map([exchange_id], |row| {
            Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))
        })?;
        rows.collect::<std::result::Result<std::collections::HashMap<_, _>, _>>()
            .context("Failed to load api aliases")
    }

    /// Load `PairsParams` from DB. Convenience wrapper over `load_pairs_params_rows`.
    pub fn load_pairs_params(&self) -> arb_common::config::PairsParams {
        match self.load_pairs_params_rows() {
            Ok(rows) => arb_common::config::PairsParams::from_rows(rows),
            Err(e) => {
                tracing::warn!(error = %e, "Failed to load pairs params from DB, using defaults");
                arb_common::config::PairsParams::default()
            }
        }
    }

    /// Load market params rows suitable for `PairsParams::from_rows()`.
    /// Returns (symbol, exchange_name, SymbolMarketRow) tuples.
    pub fn load_pairs_params_rows(&self) -> Result<Vec<(String, String, arb_common::config::SymbolMarketRow)>> {
        let mut stmt = self.conn.prepare(
            "SELECT es.symbol, e.name,
                    es.max_leverage, es.mmf, es.size_decimals, es.qty_step, es.min_qty
             FROM exchange_symbols es
             JOIN exchanges e ON e.id = es.exchange_id"
        )?;
        let rows = stmt.query_map([], |row| {
            let symbol: String = row.get(0)?;
            let exchange_name: String = row.get(1)?;
            let qty_step_str: String = row.get::<_, String>(5).unwrap_or_default();
            let min_qty_str: String = row.get::<_, String>(6).unwrap_or_default();
            Ok((symbol, exchange_name, arb_common::config::SymbolMarketRow {
                max_leverage: row.get(2)?,
                mmf: row.get(3)?,
                size_decimals: row.get::<_, Option<i32>>(4).ok().flatten().map(|v| v as u32),
                qty_step: qty_step_str.parse::<f64>().ok(),
                min_qty: min_qty_str.parse::<f64>().ok(),
            }))
        })?;
        rows.collect::<std::result::Result<Vec<_>, _>>()
            .context("Failed to load pairs params rows")
    }

    /// Apply mmf overrides from exchange_symbol_overrides to exchange_symbols.
    /// Called after API data is upserted so operator values take precedence.
    pub fn apply_mmf_overrides(&self) -> Result<usize> {
        let count = self.conn.execute(
            "UPDATE exchange_symbols SET mmf = CAST(o.value AS REAL)
             FROM exchange_symbol_overrides o
             WHERE exchange_symbols.symbol = o.symbol
               AND exchange_symbols.exchange_id = o.exchange_id
               AND o.field_name = 'mmf'",
            [],
        )?;
        Ok(count)
    }

    // ── Transaction batching ─────────────────────────────────────────────────

    /// Run multiple write operations inside a single SQLite transaction.
    /// Commits if the closure returns `Ok`, rolls back on `Err` or panic.
    ///
    /// Used by hot-path entry/close DB writes (`spawn_blocking` task)
    /// to amortize fsync cost across multiple INSERTs and to give
    /// crash-atomicity for the trade/orders/signal batch.
    ///
    /// Uses `BEGIN IMMEDIATE` so the write lock is acquired at BEGIN time,
    /// where `busy_timeout` applies. With `BEGIN DEFERRED` (the default),
    /// lock upgrades mid-transaction bypass `busy_timeout` entirely — SQLite
    /// returns SQLITE_BUSY immediately, causing lost writes.
    pub fn with_transaction<F, R>(&self, f: F) -> Result<R>
    where
        F: FnOnce(&Self) -> Result<R>,
    {
        let tx = rusqlite::Transaction::new_unchecked(
            &self.conn,
            rusqlite::TransactionBehavior::Immediate,
        ).context("BEGIN IMMEDIATE failed")?;
        let result = f(self);
        match result {
            Ok(r) => {
                tx.commit().context("COMMIT failed")?;
                Ok(r)
            }
            Err(e) => {
                // Transaction guard rolls back on drop, but be explicit so
                // failures in rollback are surfaced to logs.
                if let Err(rb) = tx.rollback() {
                    tracing::warn!(error = %rb, "Transaction rollback failed");
                }
                Err(e)
            }
        }
    }

    // ── Orders (persistent lifecycle) ────────────────────────────────────────

    pub fn insert_order(&self, o: &OrderRecord) -> Result<u32> {
        // Compute next session_seq: max seq within the same session + 1
        let session_seq: u32 = self.conn.query_row(
            "SELECT COALESCE(MAX(o.session_seq), 0) + 1
             FROM orders o JOIN trades t ON o.trade_id = t.id
             WHERE t.session_id = (SELECT session_id FROM trades WHERE id = ?1)",
            params![o.trade_id],
            |r| r.get(0),
        ).unwrap_or(1);
        self.conn.execute(
            "INSERT OR REPLACE INTO orders (
                order_id, trade_id, session_seq, symbol, exchange_id, side,
                order_type, purpose, qty, limit_price, trigger_price, status,
                placed_at_ms, exchange_ts_ms, recv_ts_ms, signal_price
            ) VALUES (?1,?2,?3,?4,?5,?6,?7,?8,?9,?10,?11,'placed',?12,?13,?14,?15)",
            params![
                o.order_id,
                o.trade_id,
                session_seq,
                o.symbol,
                o.exchange_id,
                o.side,
                o.order_type,
                o.purpose,
                dec(o.qty),
                dec(o.limit_price),
                dec(o.trigger_price),
                o.placed_at_ms,
                o.exchange_ts_ms,
                o.recv_ts_ms,
                dec(o.signal_price),
            ],
        )?;
        Ok(session_seq)
    }

    /// Update order with fill data. Handles first vs subsequent fills.
    pub fn update_order_fill(
        &self,
        order_id: u64,
        filled_qty: Decimal,
        avg_price: Decimal,
        fill_at_ms: u64,
        is_complete: bool,
        fee: Decimal,
    ) -> Result<()> {
        let status = if is_complete { "filled" } else { "partial" };
        let closed_at = if is_complete { fill_at_ms } else { 0 };
        self.conn.execute(
            "UPDATE orders SET
                filled_qty = ?1, avg_fill_price = ?2,
                first_fill_at_ms = CASE WHEN first_fill_at_ms = 0 THEN ?3 ELSE first_fill_at_ms END,
                last_fill_at_ms = ?3,
                fee = ?4, status = ?5, closed_at_ms = ?6
            WHERE order_id = ?7",
            params![
                dec(filled_qty),
                dec(avg_price),
                fill_at_ms,
                dec(fee),
                status,
                closed_at,
                order_id,
            ],
        )?;
        Ok(())
    }

    /// Update fee for an order with actual value from exchange REST API.
    pub fn update_order_fee(&self, order_id: u64, fee: Decimal) -> Result<()> {
        self.conn.execute(
            "UPDATE orders SET fee = ?1 WHERE order_id = ?2",
            params![dec(fee), order_id],
        )?;
        Ok(())
    }

    pub fn update_order_status(&self, order_id: u64, status: &str, closed_at_ms: u64, close_reason: &str) -> Result<()> {
        self.conn.execute(
            "UPDATE orders SET status = ?1,
                closed_at_ms = CASE WHEN closed_at_ms > 0 THEN closed_at_ms ELSE ?2 END,
                close_reason = ?3
            WHERE order_id = ?4",
            params![status, closed_at_ms, close_reason, order_id],
        )?;
        Ok(())
    }

    /// Store exchange-side identifiers for an order (tx_hash from Lighter, order_id from Extended).
    pub fn set_order_exchange_id(&self, order_id: u64, tx_hash: Option<&str>, exchange_order_id: Option<&str>) -> Result<()> {
        if let Some(h) = tx_hash {
            self.conn.execute(
                "UPDATE orders SET tx_hash = ?1 WHERE order_id = ?2",
                params![h, order_id],
            )?;
        }
        if let Some(eid) = exchange_order_id {
            self.conn.execute(
                "UPDATE orders SET exchange_order_id = ?1 WHERE order_id = ?2",
                params![eid, order_id],
            )?;
        }
        Ok(())
    }

    /// Persist a fill that happened on an external leg (liquidation, external SL,
    /// manual close from the exchange UI). External-leg fills never flow through
    /// the normal `track_order` → `place_order` → WS-fill pipeline, so without
    /// this helper the `orders` table would be missing the close row for that
    /// leg and `derive_pnl` would report `both_filled=false` → gross=0 (the
    /// 2026-04-18 MEGA liquidation bug: DB had entry_1, entry_2, sl_*, and only
    /// `exit_unwind_2` for the surviving leg; `exit_unwind_1` for the
    /// liquidated leg was missing → UI showed ~$0 for a real −$1.46 loss).
    ///
    /// Inserts a filled `orders` row in one shot, wrapped in a transaction
    /// so partial state (order row without fill data) can't survive a
    /// crash mid-way. The matching trade must already exist (FK).
    ///
    /// `purpose` should be `exit_unwind_1` or `exit_unwind_2`.
    /// `close_reason` should be `"liquidation"`, `"external_sl"`, or
    /// `"external_manual"`.
    #[allow(clippy::too_many_arguments)]
    pub fn persist_external_leg_fill(
        &self,
        trade_id: &str,
        order_id: u64,
        symbol: &str,
        exchange_id: i64,
        purpose: &str,
        side: &str,
        qty: Decimal,
        fill_price: Decimal,
        fee: Decimal,
        fill_at_ms: u64,
        close_reason: &str,
    ) -> Result<()> {
        self.with_transaction(|db| {
            db.persist_external_leg_fill_inner(
                trade_id, order_id, symbol, exchange_id, purpose, side,
                qty, fill_price, fee, fill_at_ms, close_reason,
            )
        })
    }

    /// Transaction-less variant of `persist_external_leg_fill`. Used by
    /// callers that are already inside a `with_transaction` block (rusqlite
    /// rejects nested BEGIN IMMEDIATE). All the actual INSERT/UPDATE
    /// statements live here; the public wrapper adds the transaction.
    #[allow(clippy::too_many_arguments)]
    fn persist_external_leg_fill_inner(
        &self,
        trade_id: &str,
        order_id: u64,
        symbol: &str,
        exchange_id: i64,
        purpose: &str,
        side: &str,
        qty: Decimal,
        fill_price: Decimal,
        fee: Decimal,
        fill_at_ms: u64,
        close_reason: &str,
    ) -> Result<()> {
        let rec = OrderRecord {
            order_id,
            trade_id: trade_id.to_string(),
            symbol: symbol.to_string(),
            exchange_id,
            side: side.to_string(),
            order_type: "ioc".to_string(),
            purpose: purpose.to_string(),
            qty,
            // The exchange did the matching — we record the observed fill
            // price as both the "limit" (synthetic) and signal_price.
            limit_price: fill_price,
            trigger_price: Decimal::ZERO,
            placed_at_ms: fill_at_ms,
            exchange_ts_ms: fill_at_ms,
            recv_ts_ms: fill_at_ms,
            signal_price: fill_price,
        };
        self.insert_order(&rec)?;
        self.update_order_fill(order_id, qty, fill_price, fill_at_ms, true, fee)?;
        self.update_order_status(order_id, "filled", fill_at_ms, close_reason)?;
        Ok(())
    }

    /// Record the equity of one exchange at session start. Callers invoke
    /// once per exchange right after `Initial balances fetched` at bot
    /// startup. `INSERT OR IGNORE` — the first snapshot for a
    /// `(session_id, exchange_id)` wins; subsequent bot restarts within
    /// the same session do not overwrite it, so the baseline against
    /// which `Balance Δ` is computed stays stable across restarts.
    pub fn insert_session_start_balance(
        &self,
        session_id: i64,
        exchange_id: i64,
        balance_usd: Decimal,
        captured_at_ms: u64,
    ) -> Result<()> {
        self.conn.execute(
            "INSERT OR IGNORE INTO session_start_balances
                (session_id, exchange_id, balance_usd, captured_at_ms)
             VALUES (?1, ?2, ?3, ?4)",
            params![session_id, exchange_id, dec(balance_usd), captured_at_ms],
        )?;
        Ok(())
    }

    /// Load per-exchange start-of-session equity for the given session.
    /// Returns `(exchange_id, balance_usd)` pairs ordered by exchange_id.
    /// Empty vec if the session pre-dates the feature.
    pub fn load_session_start_balances(
        &self,
        session_id: i64,
    ) -> Result<Vec<(i64, Decimal)>> {
        let mut stmt = self.conn.prepare(
            "SELECT exchange_id, balance_usd
             FROM session_start_balances
             WHERE session_id = ?1
             ORDER BY exchange_id",
        )?;
        let rows = stmt.query_map(params![session_id], |r| {
            let eid: i64 = r.get(0)?;
            let bal_s: String = r.get(1)?;
            let bal = Decimal::from_str(bal_s.trim()).unwrap_or(Decimal::ZERO);
            Ok((eid, bal))
        })?;
        let mut out = Vec::new();
        for row in rows { out.push(row?); }
        Ok(out)
    }

    /// Record a reconciliation-time close as an `orders` row on an existing
    /// trade. Used for Direction-mismatch and ORPHAN paths in
    /// `main.rs` reconciliation where a DB trade exists but neither leg
    /// had an `exit_%` order written (bot was killed mid-position or the
    /// exchange closed our leg out-of-band before we recorded anything).
    ///
    /// `trade_id` must already exist in the `trades` table.
    /// `purpose` is `"exit_unwind_1"` or `"exit_unwind_2"` per which leg
    /// we closed in reconciliation.
    /// `close_reason` is a short tag for the UI (`"reconcile_orphan"`,
    /// `"reconcile_dir_mismatch"`).
    ///
    /// Lower-level single-leg primitive: does NOT touch trade `state`.
    /// Production callers should use `finalize_reconciliation_close`
    /// instead, which bundles 0/1/2 leg inserts + the state flip in one
    /// transaction. Kept for the test that verifies the `orders` row
    /// shape independent of state transitions.
    #[allow(clippy::too_many_arguments)]
    pub fn record_reconciliation_exit(
        &self,
        trade_id: &str,
        order_id: u64,
        symbol: &str,
        exchange_id: i64,
        purpose: &str,
        side: &str,
        qty: Decimal,
        fill_price: Decimal,
        fee: Decimal,
        closed_at_ms: u64,
        close_reason: &str,
    ) -> Result<()> {
        // Re-use the same wire path as external-leg fills — both are
        // "filled synchronously from outside the normal IOC pipeline".
        // `persist_external_leg_fill` wraps its 3 writes in a transaction.
        self.persist_external_leg_fill(
            trade_id, order_id, symbol, exchange_id, purpose, side,
            qty, fill_price, fee, closed_at_ms, close_reason,
        )
    }

    /// Atomic finalize of a reconciliation close: persist 0/1/2 leg fills
    /// AND flip the trade `state` to `closed` in one transaction. Prevents
    /// the 2026-04-18 split where `update_trade_state_by_symbol` fired
    /// unconditionally after `close_position` → a failed close still
    /// marked the trade closed in DB, leaving an orphan on the exchange.
    ///
    /// Callers pass `legs` as `Option<ReconcileLegFill>` for each of the
    /// two exchange legs (None = leg already gone / no fill to record).
    /// GHOST path (both None) → only the state flip runs.
    pub fn finalize_reconciliation_close(
        &self,
        trade_id: &str,
        symbol: &str,
        leg1: Option<ReconcileLegFill>,
        leg2: Option<ReconcileLegFill>,
        close_reason: &str,
        closed_at_ms: u64,
    ) -> Result<()> {
        self.with_transaction(|db| {
            for leg in [leg1.as_ref(), leg2.as_ref()].into_iter().flatten() {
                db.persist_external_leg_fill_inner(
                    trade_id,
                    leg.order_id,
                    symbol,
                    leg.exchange_id,
                    &leg.purpose,
                    &leg.side,
                    leg.qty,
                    leg.fill_price,
                    leg.fee,
                    closed_at_ms,
                    close_reason,
                )?;
            }
            db.conn.execute(
                "UPDATE trades SET state = 'closed', closed_at_ms = ?1
                 WHERE symbol = ?2 AND state IN ('active', 'closing', 'filling')",
                params![closed_at_ms as i64, symbol],
            )?;
            Ok(())
        })
    }

    /// Record a synthetic "orphan cover" trade for the reconciliation LEAK
    /// path — an exchange position that has no corresponding DB trade
    /// (e.g. our paired leg was lost previously or the position was
    /// opened manually). Without this we close the orphan via
    /// `close_position`, the cost evaporates (trait returns `()`), and
    /// the UI can't see the expense (the 2026-04-18 HL naked short
    /// bleed during the MEGA price spike).
    ///
    /// Creates:
    ///   * one `trades` row with `state='closed'`, `exit_reason='orphan_cover'`
    ///   * one `entry_1` order on `orphan_exchange_id` @ `entry_price`
    ///     (the position's observed entry price from `fetch_open_positions`)
    ///   * one `entry_2` + one `exit_unwind_2` order on
    ///     `hedge_exchange_id` @ `entry_price` — a synthetic "hedge" leg
    ///     whose entry==exit so `(x2-e2)*q = 0` contributes nothing to
    ///     `derive_pnl`; it exists purely to keep the AGG_TRADES_SELECT
    ///     LEFT JOIN deterministic. Pass a different exchange ID from
    ///     `orphan_exchange_id` to avoid cartesian-product duplication
    ///     in the aggregation query.
    ///   * one `exit_unwind_1` order on `orphan_exchange_id` @ `cover_price`
    ///     — the actual buy-back / sell-back that matters.
    ///
    /// Returns the synthetic `trade_id`.
    #[allow(clippy::too_many_arguments)]
    pub fn record_orphan_cover_trade(
        &self,
        session_id: i64,
        symbol: &str,
        orphan_exchange_id: i64,
        hedge_exchange_id: i64,
        qty: Decimal,
        was_long: bool,
        entry_price: Decimal,
        cover_price: Decimal,
        cover_fee: Decimal,
        closed_at_ms: u64,
        orphan_order_id: u64,
        hedge_entry_order_id: u64,
        hedge_exit_order_id: u64,
        orphan_exit_order_id: u64,
    ) -> Result<String> {
        if orphan_exchange_id == hedge_exchange_id {
            // Same-exchange fallback — the aggregation JOIN will duplicate
            // rows. This only happens in pathological configs (single
            // exchange), production runs 4. Warn so the operator can spot.
            tracing::warn!(
                symbol, exchange_id = orphan_exchange_id,
                "record_orphan_cover_trade: hedge_exchange_id == orphan_exchange_id — aggregation will double-count this trade"
            );
        }
        let trade_id = uuid::Uuid::new_v4().to_string();
        // Direction: `was_long` refers to the orphan leg. Entry side:
        //   long orphan entered with a BUY, closed by SELL
        //   short orphan entered with a SELL, closed by BUY
        // For the synthetic hedge leg we use the opposite side (as if it
        // were the other leg of an arb) but since e2==x2 the direction
        // doesn't affect PnL.
        let orphan_entry_side = if was_long { "buy" } else { "sell" };
        let orphan_exit_side  = if was_long { "sell" } else { "buy" };
        let hedge_entry_side  = if was_long { "sell" } else { "buy" };
        let hedge_exit_side   = if was_long { "buy" } else { "sell" };

        // Wrap the trade + 4 orders in a single transaction. Partial
        // state (trade row without all orders, or some orders with
        // `status='placed'` because the follow-up UPDATE failed) would
        // silently poison `derive_pnl` — gross falls back to 0 on any
        // missing/unfilled leg and the UI loses the expense.
        let tid_ref = &trade_id;
        self.with_transaction(|db| {
            db.conn.execute(
                "INSERT INTO trades (
                    id, session_id, session_seq, symbol, is_long_1,
                    exchange_id_1, exchange_id_2, qty, state,
                    opened_at_ms, closed_at_ms, signal_at_ms, exit_reason,
                    entry_spread_bps, funding_1, funding_2
                 ) VALUES (?1, ?2, 1, ?3, ?4, ?5, ?6, ?7, 'closed',
                           ?8, ?8, ?8, 'orphan_cover',
                           '0', '0', '0')",
                params![
                    tid_ref, session_id, symbol, was_long as i64,
                    orphan_exchange_id, hedge_exchange_id, dec(qty),
                    closed_at_ms,
                ],
            )?;
            // Entry on orphan leg (what we observed on the exchange).
            db.persist_external_leg_fill_inner(
                tid_ref, orphan_order_id, symbol, orphan_exchange_id,
                "entry_1", orphan_entry_side, qty, entry_price, Decimal::ZERO,
                closed_at_ms, "orphan_observed_entry",
            )?;
            // Synthetic hedge entry + exit — entry==exit so contributes 0 PnL.
            db.persist_external_leg_fill_inner(
                tid_ref, hedge_entry_order_id, symbol, hedge_exchange_id,
                "entry_2", hedge_entry_side, qty, entry_price, Decimal::ZERO,
                closed_at_ms, "orphan_synthetic_hedge",
            )?;
            db.persist_external_leg_fill_inner(
                tid_ref, hedge_exit_order_id, symbol, hedge_exchange_id,
                "exit_unwind_2", hedge_exit_side, qty, entry_price, Decimal::ZERO,
                closed_at_ms, "orphan_synthetic_hedge",
            )?;
            // Real exit: the cover fill that actually moved money.
            db.persist_external_leg_fill_inner(
                tid_ref, orphan_exit_order_id, symbol, orphan_exchange_id,
                "exit_unwind_1", orphan_exit_side, qty, cover_price, cover_fee,
                closed_at_ms, "orphan_cover",
            )?;
            Ok(())
        })?;
        Ok(trade_id)
    }

    /// Update an SL order after a successful move (drift or renewal).
    /// Lookup by trade_id + purpose (stable key), updates order_id (may change on Extended),
    /// trigger/limit prices, and increments move_count.
    pub fn update_sl_after_move(
        &self,
        trade_id: &str,
        purpose: &str,
        new_order_id: u64,
        trigger_price: &str,
        limit_price: &str,
        updated_at_ms: u64,
    ) -> Result<()> {
        self.conn.execute(
            "UPDATE orders SET order_id = ?1, exchange_order_id = ?1, trigger_price = ?2, limit_price = ?3,
                updated_at_ms = ?4, move_count = move_count + 1
            WHERE trade_id = ?5 AND purpose = ?6 AND status = 'placed'",
            params![new_order_id, trigger_price, limit_price, updated_at_ms, trade_id, purpose],
        )?;
        Ok(())
    }

    /// Cancel all non-terminal orders on startup (clean mode — everything is wiped).
    pub fn cancel_stale_orders(&self) -> Result<u64> {
        let now_ms = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap()
            .as_millis() as u64;
        let count = self.conn.execute(
            "UPDATE orders SET status = 'cancelled', closed_at_ms = ?1, close_reason = 'bot restarted'
             WHERE status IN ('placed', 'partial')",
            params![now_ms],
        )?;
        Ok(count as u64)
    }

    /// Cancel only non-SL stale orders (resume mode — SL orders are preserved on exchanges).
    pub fn cancel_non_sl_orders(&self) -> Result<u64> {
        let now_ms = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap()
            .as_millis() as u64;
        let count = self.conn.execute(
            "UPDATE orders SET status = 'cancelled', closed_at_ms = ?1, close_reason = 'bot restarted'
             WHERE status IN ('placed', 'partial') AND purpose NOT LIKE 'sl_%'",
            params![now_ms],
        )?;
        Ok(count as u64)
    }

    /// Load active SL orders from DB (status='placed') for resume into active_orders DashMap.
    pub fn load_active_sl_orders(&self) -> Result<Vec<ActiveSlOrder>> {
        let mut stmt = self.conn.prepare(
            "SELECT order_id, session_seq, trade_id, symbol, exchange_id, side, order_type, purpose, qty, limit_price, trigger_price, placed_at_ms, exchange_ts_ms, recv_ts_ms, exchange_order_id
             FROM orders
             WHERE status = 'placed' AND purpose LIKE 'sl_%'"
        )?;
        let rows = stmt.query_map([], |row| {
            let exchange_order_id_str: Option<String> = row.get::<_, Option<String>>(14).unwrap_or(None);
            let exchange_order_id = exchange_order_id_str.and_then(|s| s.parse::<u64>().ok());
            Ok(ActiveSlOrder {
                order_id: row.get(0)?,
                session_seq: row.get::<_, u32>(1).unwrap_or(0),
                trade_id: row.get(2)?,
                symbol: row.get(3)?,
                exchange_id: row.get(4)?,
                side: row.get(5)?,
                order_type: row.get(6)?,
                purpose: row.get(7)?,
                qty: row.get(8)?,
                limit_price: row.get(9)?,
                trigger_price: row.get::<_, String>(10).unwrap_or_else(|_| "0".to_string()),
                placed_at_ms: row.get(11)?,
                exchange_ts_ms: row.get::<_, Option<u64>>(12)?.unwrap_or(0),
                recv_ts_ms: row.get::<_, Option<u64>>(13)?.unwrap_or(0),
                exchange_order_id,
            })
        })?.filter_map(|r| r.ok()).collect();
        Ok(rows)
    }

    /// Load stale orders for a specific exchange (status='placed'/'partial') for REST recovery.
    /// Returns (order_id, client_order_id_str, symbol).
    pub fn load_stale_orders_for_exchange(&self, exchange_id: i64) -> Result<Vec<(u64, String, String)>> {
        let mut stmt = self.conn.prepare(
            "SELECT order_id, symbol FROM orders
             WHERE exchange_id = ?1 AND status IN ('placed', 'partial')
             ORDER BY placed_at_ms DESC LIMIT 100"
        )?;
        let rows = stmt.query_map(params![exchange_id], |row| {
            let order_id: u64 = row.get(0)?;
            let symbol: String = row.get(1)?;
            Ok((order_id, order_id.to_string(), symbol))
        })?.filter_map(|r| r.ok()).collect();
        Ok(rows)
    }

    /// Fetch all orders linked to a trade (entry, SL, close).
    pub fn query_orders_for_trade(&self, trade_id: &str) -> Result<Vec<OrderRow>> {
        let mut stmt = self.conn.prepare(
            "SELECT o.order_id, o.exchange_id, o.side, o.order_type, o.purpose, o.qty, o.limit_price,
                    o.status, o.filled_qty, o.avg_fill_price, o.fee, o.placed_at_ms, o.first_fill_at_ms,
                    o.exchange_ts_ms, o.recv_ts_ms, o.signal_price,
                    CASE WHEN o.purpose LIKE 'exit%'
                         THEN t.exit_signal_at_ms ELSE t.signal_at_ms END
             FROM orders o
             JOIN trades t ON o.trade_id = t.id
             WHERE o.trade_id = ?1 AND (o.purpose NOT LIKE 'sl_%' OR o.status = 'filled')
             ORDER BY o.placed_at_ms ASC"
        )?;
        let rows = stmt.query_map(params![trade_id], |row| {
            let qty_str: String = row.get(5)?;
            let price_str: String = row.get(6)?;
            Ok(OrderRow {
                order_id: row.get(0)?,
                exchange_id: row.get(1)?,
                side: row.get(2)?,
                order_type: row.get(3)?,
                purpose: row.get(4)?,
                qty: qty_str.parse().unwrap_or_default(),
                limit_price: price_str.parse().unwrap_or_default(),
                status: row.get(7)?,
                filled_qty: row.get::<_, String>(8).unwrap_or_default(),
                avg_fill_price: row.get::<_, String>(9).unwrap_or_default(),
                fee: row.get::<_, String>(10).unwrap_or_default(),
                placed_at_ms: row.get::<_, u64>(11).unwrap_or(0),
                first_fill_at_ms: row.get::<_, u64>(12).unwrap_or(0),
                exchange_ts_ms: row.get::<_, u64>(13).unwrap_or(0),
                recv_ts_ms: row.get::<_, u64>(14).unwrap_or(0),
                signal_price: row.get::<_, String>(15).unwrap_or_default(),
                signal_at_ms: row.get::<_, u64>(16).unwrap_or(0),
            })
        })?.filter_map(|r| r.ok()).collect();
        Ok(rows)
    }

    /// Cancel all active orders linked to a specific trade (e.g. SL orders after trade closes).
    pub fn cancel_orders_for_trade(&self, trade_id: &str) -> Result<u64> {
        let now_ms = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap()
            .as_millis() as u64;
        let count = self.conn.execute(
            "UPDATE orders SET status = 'cancelled', closed_at_ms = ?1, close_reason = 'trade closed'
             WHERE trade_id = ?2 AND status IN ('placed', 'partial')",
            params![now_ms, trade_id],
        )?;
        Ok(count as u64)
    }

    // ── Trades + Legs ──────────────────────────────────────────────────────

    /// Insert a trade (or update if it already exists, preserving session_id/session_seq).
    /// Returns the session_seq assigned to this trade.
    pub fn insert_trade(&self, t: &TradeRecord) -> Result<i64> {
        // Compute session_id + session_seq for new trades
        let session_id: i64 = self.conn.query_row(
            "SELECT id FROM sessions WHERE state = 'active' ORDER BY id DESC LIMIT 1",
            [], |r| r.get(0),
        ).unwrap_or(0);
        let next_seq: i64 = self.conn.query_row(
            "SELECT COALESCE(MAX(session_seq), 0) + 1 FROM trades WHERE session_id = ?1",
            params![session_id], |r| r.get(0),
        ).unwrap_or(1);

        self.conn.execute(
            "INSERT INTO trades (
                id, symbol, is_long_1, exchange_id_1, exchange_id_2,
                qty, state, opened_at_ms,
                session_id, session_seq,
                signal_at_ms, entry_spread_bps,
                cfg_entry_bps, cfg_include_fees_in_entry,
                cfg_exit_mode, cfg_exit_bps, cfg_exit_pnl_pct, cfg_include_fees_in_exit,
                cfg_leverage
            ) VALUES (?1,?2,?3,?4,?5,?6,?7,?8,?9,?10,?11,?12,?13,?14,?15,?16,?17,?18,?19)
            ON CONFLICT(id) DO UPDATE SET
                symbol=excluded.symbol, is_long_1=excluded.is_long_1, qty=excluded.qty,
                state=excluded.state, opened_at_ms=excluded.opened_at_ms,
                entry_spread_bps=excluded.entry_spread_bps,
                cfg_entry_bps=excluded.cfg_entry_bps,
                cfg_include_fees_in_entry=excluded.cfg_include_fees_in_entry,
                cfg_exit_mode=excluded.cfg_exit_mode,
                cfg_exit_bps=excluded.cfg_exit_bps,
                cfg_exit_pnl_pct=excluded.cfg_exit_pnl_pct,
                cfg_include_fees_in_exit=excluded.cfg_include_fees_in_exit,
                cfg_leverage=excluded.cfg_leverage",
            params![
                t.id, t.symbol, t.is_long_1 as i64, t.exchange_id_1, t.exchange_id_2,
                dec(t.qty), t.state, t.opened_at_ms,
                session_id, next_seq,
                t.signal_at_ms, dec(t.entry_spread_bps),
                dec(t.cfg.entry_bps), t.cfg.include_fees_in_entry as i64,
                t.cfg.exit_mode, dec(t.cfg.exit_bps), dec(t.cfg.exit_pnl_pct),
                t.cfg.include_fees_in_exit as i64,
                t.cfg.leverage as i64,
            ],
        )?;
        // Return actual session_seq (preserved on conflict, new on first insert)
        let actual_seq: i64 = self.conn.query_row(
            "SELECT COALESCE(session_seq, 0) FROM trades WHERE id = ?1",
            params![t.id], |r| r.get(0),
        ).unwrap_or(0);
        Ok(actual_seq)
    }

    pub fn update_trade_state(&self, id: &Uuid, state: &str) -> Result<()> {
        self.conn.execute(
            "UPDATE trades SET state = ?1 WHERE id = ?2",
            params![state, id.to_string()],
        )?;
        Ok(())
    }

    /// Persist the local wall-clock time (ms) when the close signal fired for
    /// this trade. Called once from `execute_close` before the close orders
    /// are placed, so the UI Recv→Sig metric for close orders has a correct
    /// per-trade reference (symmetric to `signal_at_ms` for entry).
    pub fn update_trade_exit_signal_at(&self, id: &Uuid, exit_signal_at_ms: u64) -> Result<()> {
        self.conn.execute(
            "UPDATE trades SET exit_signal_at_ms = ?1 WHERE id = ?2",
            params![exit_signal_at_ms, id.to_string()],
        )?;
        Ok(())
    }

    /// Transition a `filling` trade to `failed_open`. PnL/fees are derived at
    /// read time from `orders`; this only updates trade state + qty +
    /// lifecycle timestamps.
    pub fn update_trade_as_failed_open(
        &self, trade_id: &str,
        qty: Decimal,
        opened_at_ms: u64, closed_at_ms: u64,
    ) -> Result<()> {
        self.conn.execute(
            "UPDATE trades SET state = 'failed_open', qty = ?1,
                opened_at_ms = ?2, closed_at_ms = ?3
             WHERE id = ?4",
            params![dec(qty), opened_at_ms, closed_at_ms, trade_id],
        )?;
        Ok(())
    }

    /// Insert a row into the `signals` audit table.
    ///
    /// Single INSERT, runs in microseconds. Callers should invoke from
    /// `tokio::task::spawn_blocking` to keep the async hot path clear.
    pub fn insert_signal(&self, row: &crate::storage::types::SignalRow) -> Result<()> {
        self.conn.execute(
            "INSERT INTO signals (
                id, session_id, ts_ms, signal_type, symbol,
                spread_bps, staleness_ms,
                threshold_bps, fee_deduct_bps, order_qty, order_qty_usd,
                exit_reason, entry_spread_bps, gross_pnl_bps, net_pnl_pct,
                held_for_ms, funding_at_exit,
                legs, trade_id, rejected_reason
            ) VALUES (
                ?1, ?2, ?3, ?4, ?5,
                ?6, ?7,
                ?8, ?9, ?10, ?11,
                ?12, ?13, ?14, ?15,
                ?16, ?17,
                ?18, ?19, ?20
            )",
            params![
                row.id,
                row.session_id,
                row.ts_ms,
                row.signal_type,
                row.symbol,
                dec(row.spread_bps),
                row.staleness_ms,
                row.threshold_bps.map(dec),
                row.fee_deduct_bps.map(dec),
                row.order_qty.map(dec),
                row.order_qty_usd.map(dec),
                row.exit_reason,
                row.entry_spread_bps.map(dec),
                row.gross_pnl_bps.map(dec),
                row.net_pnl_pct.map(dec),
                row.held_for_ms,
                row.funding_at_exit.map(dec),
                row.legs_json,
                row.trade_id,
                row.rejected_reason,
            ],
        )?;
        Ok(())
    }

    /// Update cumulative funding for an active trade (called periodically from `housekeeping_tick`).
    pub fn update_trade_funding(&self, id: &str, funding_1: &str, funding_2: &str) -> Result<()> {
        self.conn.execute(
            "UPDATE trades SET funding_1 = ?1, funding_2 = ?2
             WHERE id = ?3 AND state NOT IN ('closed','failed_open','closed_error')",
            params![funding_1, funding_2, id],
        )?;
        Ok(())
    }

    /// Close a trade: update `trades` row with close timestamp, exit reason,
    /// hot-reloadable exit config snapshot, and final funding.
    ///
    /// **No PnL/fees written here.** Gross PnL, fees and net PnL are all
    /// derived at read time from `orders` + this row's structural fields —
    /// see `load_*_closed_trades` queries and the ClosedTradeRow constructor
    /// in this file.
    pub fn close_trade(
        &self,
        id: &Uuid,
        exit_reason: &str,
        closed_at_ms: u64,
        exit_mode: &str,
        exit_bps: Decimal,
        exit_pnl_pct: Decimal,
        include_fees_in_exit: bool,
        funding_1: &str,
        funding_2: &str,
    ) -> Result<()> {
        self.conn.execute(
            "UPDATE trades SET
                state = 'closed',
                exit_reason = ?1, closed_at_ms = ?2,
                cfg_exit_mode = ?3, cfg_exit_bps = ?4, cfg_exit_pnl_pct = ?5,
                cfg_include_fees_in_exit = ?6,
                funding_1 = ?7, funding_2 = ?8
            WHERE id = ?9",
            params![
                exit_reason, closed_at_ms,
                exit_mode, dec(exit_bps), dec(exit_pnl_pct),
                include_fees_in_exit as i64,
                funding_1, funding_2,
                id.to_string(),
            ],
        )?;
        Ok(())
    }

    /// Update exit params for a single trade. Used by hot-config reload to
    /// rewrite each active trade's persisted cfg with its effective snapshot
    /// (per-symbol overrides applied) — see executor's hot-reload path.
    pub fn update_trade_exit_cfg(
        &self,
        id: &str,
        exit_mode: &str,
        exit_bps: Decimal,
        exit_pnl_pct: Decimal,
        include_fees_in_exit: bool,
    ) -> Result<()> {
        self.conn.execute(
            "UPDATE trades SET
                cfg_exit_mode = ?1, cfg_exit_bps = ?2, cfg_exit_pnl_pct = ?3,
                cfg_include_fees_in_exit = ?4
             WHERE id = ?5",
            params![
                exit_mode, dec(exit_bps), dec(exit_pnl_pct),
                include_fees_in_exit as i64,
                id,
            ],
        )?;
        Ok(())
    }

    pub fn close_all_active_trades(&self, closed_at_ms: i64) -> Result<u64> {
        let n = self.conn.execute(
            "UPDATE trades SET state = 'closed', closed_at_ms = ?1, exit_reason = 'clean_restart'
             WHERE state IN ('active', 'closing', 'filling')",
            params![closed_at_ms],
        )?;
        Ok(n as u64)
    }

    /// Update trade state for a specific symbol.
    /// Reconciliation callers should use `finalize_reconciliation_close`
    /// instead (does the UPDATE atomically with order-row inserts).
    /// Kept as a standalone primitive for other callers / potential
    /// future admin paths.
    pub fn update_trade_state_by_symbol(&self, symbol: &str, state: &str, closed_at_ms: i64) -> Result<u64> {
        let n = self.conn.execute(
            "UPDATE trades SET state = ?1, closed_at_ms = ?2
             WHERE symbol = ?3 AND state IN ('active', 'closing', 'filling')",
            params![state, closed_at_ms, symbol],
        )?;
        Ok(n as u64)
    }

    pub fn insert_failed_open(
        &self, trade_id: &str, symbol: &str, is_long_1: bool,
        exchange_id_1: i64, exchange_id_2: i64,
        qty: Decimal, _entry_price: Decimal,
        opened_at_ms: u64, closed_at_ms: u64,
        signal_at_ms: u64,
    ) -> Result<()> {
        // Compute session_id + session_seq
        let session_id: i64 = self.conn.query_row(
            "SELECT id FROM sessions WHERE state = 'active' ORDER BY id DESC LIMIT 1",
            [], |r| r.get(0),
        ).unwrap_or(0);
        let session_seq: i64 = self.conn.query_row(
            "SELECT COALESCE(MAX(session_seq), 0) + 1 FROM trades WHERE session_id = ?1",
            params![session_id], |r| r.get(0),
        ).unwrap_or(1);

        self.conn.execute(
            "INSERT INTO trades (
                id, symbol, is_long_1, exchange_id_1, exchange_id_2,
                qty, state,
                opened_at_ms, closed_at_ms, session_id, session_seq,
                signal_at_ms
            ) VALUES (?1,?2,?3,?4,?5,?6,'failed_open',?7,?8,?9,?10,?11)",
            params![
                trade_id, symbol, is_long_1 as i64, exchange_id_1, exchange_id_2,
                dec(qty),
                opened_at_ms, closed_at_ms, session_id, session_seq,
                signal_at_ms,
            ],
        )?;
        Ok(())
    }

    pub fn load_active_trades(&self) -> Result<Vec<TradeRow>> {
        let mut stmt = self.conn.prepare(
            "SELECT t.id, t.symbol, t.is_long_1, t.exchange_id_1, t.exchange_id_2,
                    t.qty, t.state, t.opened_at_ms, t.session_seq,
                    t.signal_at_ms,
                    COALESCE(o1.signal_price, '0'), COALESCE(o2.signal_price, '0'),
                    COALESCE(o1.exchange_ts_ms, 0), COALESCE(o2.exchange_ts_ms, 0),
                    COALESCE(o1.avg_fill_price, '0') AS entry_price_1,
                    COALESCE(o2.avg_fill_price, '0') AS entry_price_2,
                    COALESCE(t.funding_1, '0'), COALESCE(t.funding_2, '0'),
                    COALESCE(o1.recv_ts_ms, 0), COALESCE(o2.recv_ts_ms, 0)
             FROM trades t
             LEFT JOIN orders o1 ON o1.trade_id = t.id AND o1.exchange_id = t.exchange_id_1 AND o1.purpose LIKE 'entry_%'
             LEFT JOIN orders o2 ON o2.trade_id = t.id AND o2.exchange_id = t.exchange_id_2 AND o2.purpose LIKE 'entry_%'
             WHERE t.state IN ('active', 'closing')
             ORDER BY t.id"
        )?;
        let trades: Vec<TradeRow> = stmt.query_map([], |row| {
            let id_str: String = row.get(0)?;
            let id = Uuid::from_str(&id_str).unwrap_or_default();
            let state_str: String = row.get(6)?;
            Ok(TradeRow {
                id,
                symbol: row.get(1)?,
                is_long_1: row.get::<_, i64>(2).unwrap_or(1) != 0,
                exchange_id_1: row.get(3)?,
                exchange_id_2: row.get(4)?,
                qty: parse_decimal(&row.get::<_, String>(5)?),
                state: parse_state(&state_str),
                opened_at_ms: row.get(7)?,
                session_seq: row.get::<_, i64>(8).unwrap_or(0),
                signal_at_ms: row.get::<_, u64>(9).unwrap_or(0),
                signal_price_1: parse_decimal(&row.get::<_, String>(10).unwrap_or_default()),
                signal_price_2: parse_decimal(&row.get::<_, String>(11).unwrap_or_default()),
                tick_at_ms_1: row.get::<_, u64>(12).unwrap_or(0),
                tick_at_ms_2: row.get::<_, u64>(13).unwrap_or(0),
                entry_price_1: parse_decimal(&row.get::<_, String>(14).unwrap_or_default()),
                entry_price_2: parse_decimal(&row.get::<_, String>(15).unwrap_or_default()),
                funding_1: parse_decimal(&row.get::<_, String>(16).unwrap_or_default()),
                funding_2: parse_decimal(&row.get::<_, String>(17).unwrap_or_default()),
                recv_at_ms_1: row.get::<_, u64>(18).unwrap_or(0),
                recv_at_ms_2: row.get::<_, u64>(19).unwrap_or(0),
            })
        })?.filter_map(|r| r.ok()).collect();
        Ok(trades)
    }

    /// Load trades closed during the current session.
    pub fn load_session_closed_trades(&self) -> Result<Vec<ClosedTradeRow>> {
        let session_id: i64 = self.conn.query_row(
            "SELECT id FROM sessions WHERE state = 'active' ORDER BY id DESC LIMIT 1",
            [], |row| row.get(0),
        ).unwrap_or(0);

        let mut stmt = self.conn.prepare(CLOSED_TRADES_SELECT_SESSION)?;
        let trades: Vec<ClosedTradeRow> = stmt.query_map(params![session_id], |row| {
            Ok(closed_trade_row_from_sql(row))
        })?.filter_map(|r| r.ok()).collect();

        Ok(trades)
    }

    /// Load closed/filled/expired orders from the current session.
    /// Returns most recent first.
    pub fn load_session_closed_orders(&self) -> Result<Vec<ClosedOrderRow>> {
        let session_id: i64 = self.conn.query_row(
            "SELECT id FROM sessions WHERE state = 'active' ORDER BY id DESC LIMIT 1",
            [], |row| row.get(0),
        ).unwrap_or(0);
        self.load_closed_orders_for_session(session_id)
    }

    /// Same as `load_session_closed_orders` but for an arbitrary historical session.
    pub fn load_closed_orders_for_session(&self, session_id: i64) -> Result<Vec<ClosedOrderRow>> {
        let mut stmt = self.conn.prepare(
            "SELECT o.order_id, o.trade_id, o.session_seq, t.symbol, o.exchange_id, o.side, o.order_type, o.purpose,
                    o.qty, o.limit_price, o.status, o.filled_qty, o.avg_fill_price,
                    o.fee, o.placed_at_ms, o.first_fill_at_ms, o.closed_at_ms, o.close_reason,
                    CASE WHEN o.purpose LIKE 'exit%'
                         THEN t.exit_signal_at_ms ELSE t.signal_at_ms END,
                    o.exchange_ts_ms, o.recv_ts_ms, o.signal_price
             FROM orders o
             JOIN trades t ON o.trade_id = t.id
             WHERE o.status != 'placed'
               AND (o.purpose NOT LIKE 'sl_%' OR o.status = 'filled')
               AND t.session_id = ?1
             ORDER BY o.closed_at_ms DESC, o.placed_at_ms DESC"
        )?;
        let rows = stmt.query_map(params![session_id], |row| {
            Ok(ClosedOrderRow {
                order_id: row.get(0)?,
                trade_id: row.get(1)?,
                session_seq: row.get::<_, u32>(2).unwrap_or(0),
                symbol: row.get(3)?,
                exchange_id: row.get(4)?,
                side: row.get(5)?,
                order_type: row.get(6)?,
                purpose: row.get(7)?,
                qty: row.get(8)?,
                limit_price: row.get(9)?,
                status: row.get(10)?,
                filled_qty: row.get::<_, String>(11).unwrap_or_default(),
                avg_fill_price: row.get::<_, String>(12).unwrap_or_default(),
                fee: row.get::<_, String>(13).unwrap_or_default(),
                placed_at_ms: row.get::<_, u64>(14).unwrap_or(0),
                first_fill_at_ms: row.get::<_, u64>(15).unwrap_or(0),
                closed_at_ms: row.get::<_, i64>(16).unwrap_or(0),
                close_reason: row.get::<_, String>(17).unwrap_or_default(),
                signal_at_ms: row.get::<_, u64>(18).unwrap_or(0),
                exchange_ts_ms: row.get::<_, u64>(19).unwrap_or(0),
                recv_ts_ms: row.get::<_, u64>(20).unwrap_or(0),
                signal_price: row.get::<_, String>(21).unwrap_or_default(),
            })
        })?.filter_map(|r| r.ok()).collect();
        Ok(rows)
    }

    // ── Snapshot ──────────────────────────────────────────────────────────────

    // ── Bot runs ────────────────────────────────────────────────────────────

    pub fn log_bot_run(&self, params: &BotRunParams) -> Result<()> {
        let now = chrono::Utc::now();
        let _now_ms = now.timestamp_millis();
        let started_at = now.format("%Y-%m-%d %H:%M:%S UTC").to_string();

        // Update the early "starting" record with full config params
        self.conn.execute(
            "UPDATE bot_runs SET
                min_entry_spread_bps = ?1, min_exit_pnl_pct = ?2, exit_mode = ?3,
                leverage = ?4, max_order_size_usd = ?5, max_open_positions = ?6,
                max_positions_per_symbol = ?7, max_used_capital_usd = ?8,
                stop_loss_buffer_pct = ?9, signal_cooldown_secs = ?10
             WHERE id = (SELECT MAX(id) FROM bot_runs)",
            params![
                params.min_entry_spread_bps,
                params.min_exit_pnl_pct,
                params.exit_mode,
                params.leverage,
                params.max_order_size_usd,
                params.max_open_positions,
                params.max_positions_per_symbol,
                params.max_used_capital_usd,
                params.stop_loss_buffer_pct,
                params.signal_cooldown_secs,
            ],
        )?;

        tracing::info!(
            started_at = %started_at,
            entry_bps = params.min_entry_spread_bps,
            tp_pct = params.min_exit_pnl_pct,
            leverage = params.leverage,
            "Bot run logged to database (state=starting)"
        );
        Ok(())
    }

    // ── Sessions ─────────────────────────────────────────────────────────

    /// Ensure a session exists for the current run and link it.
    /// Clean mode: end any active session, create a new one.
    /// Resume mode: reuse the active session (or create one if none exists).
    /// Returns (session_id, session_started_at_ms).
    pub fn ensure_session(&self, is_resume: bool) -> Result<(i64, i64)> {
        if is_resume {
            // Try to find an active session
            match self.conn.query_row(
                "SELECT id, started_at_ms FROM sessions WHERE state = 'active' ORDER BY id DESC LIMIT 1",
                [],
                |r| Ok((r.get::<_, i64>(0)?, r.get::<_, i64>(1)?)),
            ) {
                Ok((id, ms)) => {
                    self.link_current_run_to_session(id)?;
                    Ok((id, ms))
                }
                Err(_) => {
                    // No active session — create one (first run or all sessions ended)
                    self.create_session()
                }
            }
        } else {
            // Clean: end active session, create new
            let ended = self.end_active_session("clean_start")?;
            if ended > 0 {
                tracing::info!(ended, "Ended previous active session(s)");
            }
            self.create_session()
        }
    }

    /// Create a new active session and link the current run to it.
    /// Returns (session_id, started_at_ms).
    fn create_session(&self) -> Result<(i64, i64)> {
        let now_ms = chrono::Utc::now().timestamp_millis();
        self.conn.execute(
            "INSERT INTO sessions (started_at_ms, state) VALUES (?1, 'active')",
            params![now_ms],
        )?;
        let session_id = self.conn.last_insert_rowid();
        self.link_current_run_to_session(session_id)?;
        Ok((session_id, now_ms))
    }

    /// End all active sessions (there should be at most one).
    /// Returns the number of sessions ended.
    pub fn end_active_session(&self, reason: &str) -> Result<u64> {
        let now_ms = chrono::Utc::now().timestamp_millis();
        let count = self.conn.execute(
            "UPDATE sessions SET state = 'ended', ended_at_ms = ?1, end_reason = ?2
             WHERE state = 'active'",
            params![now_ms, reason],
        )?;
        Ok(count as u64)
    }

    /// Return the most recently ended session's id, by `ended_at_ms`.
    /// Used by clean-mode startup to attribute residual-position closes
    /// to the session where those positions existed, rather than the new
    /// session that's just starting. Returns `None` when no ended
    /// session exists (first-ever bot run after migrate).
    pub fn load_last_ended_session(&self) -> Result<Option<i64>> {
        let res: rusqlite::Result<i64> = self.conn.query_row(
            "SELECT id FROM sessions WHERE state = 'ended'
             ORDER BY ended_at_ms DESC, id DESC LIMIT 1",
            [],
            |r| r.get(0),
        );
        match res {
            Ok(id) => Ok(Some(id)),
            Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
            Err(e) => Err(e.into()),
        }
    }

    /// Link the most recent bot_runs row to a session.
    fn link_current_run_to_session(&self, session_id: i64) -> Result<()> {
        self.conn.execute(
            "UPDATE bot_runs SET session_id = ?1 WHERE id = (SELECT MAX(id) FROM bot_runs)",
            params![session_id],
        )?;
        Ok(())
    }

    /// Update bot run state to "running" (all systems online).
    pub fn set_bot_state_running(&self) -> Result<()> {
        let now = chrono::Utc::now();
        let running_at = now.format("%Y-%m-%d %H:%M:%S UTC").to_string();
        let running_at_ms = now.timestamp_millis();
        self.conn.execute(
            "UPDATE bot_runs SET state = 'running', running_at = ?1, running_at_ms = ?2
             WHERE id = (SELECT MAX(id) FROM bot_runs)",
            params![running_at, running_at_ms],
        )?;
        Ok(())
    }

    /// Update bot run state to "stopped" (graceful shutdown).
    pub fn set_bot_state_stopped(&self) -> Result<()> {
        let now = chrono::Utc::now();
        let stopped_at = now.format("%Y-%m-%d %H:%M:%S UTC").to_string();
        let stopped_at_ms = now.timestamp_millis();
        self.conn.execute(
            "UPDATE bot_runs SET state = 'stopped', stopped_at = ?1, stopped_at_ms = ?2
             WHERE id = (SELECT MAX(id) FROM bot_runs)",
            params![stopped_at, stopped_at_ms],
        )?;
        Ok(())
    }

    // ── History (cross-session queries) ────────────────────────────────────

    /// Load all sessions with aggregated trade stats for the history page.
    ///
    /// PnL is folded through `derive_pnl` in Rust — the same path as
    /// `load_closed_trades_for_session`. There is no SQL-side PnL math, so
    /// the aggregate numbers and the per-trade drawer can never drift.
    pub fn load_all_sessions_summary(&self) -> Result<Vec<SessionSummaryRow>> {
        use std::collections::HashMap;
        use rust_decimal::Decimal;

        // Per-session aggregates: (trade_count, wins, losses, total_pnl, total_fees, total_funding).
        // wins = net > 0, losses = net <= 0 — matches Telegram convention
        // in telegram/worker.rs:166 and the old SQL at sqlite.rs:1138-1139.
        // Partial-fill trades (failed_open with one leg) land with
        // gross = 0 from derive_pnl → net = -fees + funding, which
        // almost always counts as a loss. Do NOT introduce a
        // "neither win nor loss" bucket — it is a UI breaking change.
        let mut agg: HashMap<i64, (i64, i64, i64, Decimal, Decimal, Decimal)> = HashMap::new();
        for t in self.load_trades_for_aggregation(None)? {
            let (_gross, net, _pct) = derive_pnl(
                t.is_long_1, &t.qty,
                &t.entry_1, &t.entry_2, &t.exit_1, &t.exit_2,
                &t.fees, &t.funding_1, &t.funding_2,
            );
            let fees = Decimal::from_str(t.fees.trim()).unwrap_or(Decimal::ZERO);
            let f1 = Decimal::from_str(t.funding_1.trim()).unwrap_or(Decimal::ZERO);
            let f2 = Decimal::from_str(t.funding_2.trim()).unwrap_or(Decimal::ZERO);
            let slot = agg.entry(t.session_id).or_insert((0, 0, 0, Decimal::ZERO, Decimal::ZERO, Decimal::ZERO));
            slot.0 += 1;
            if net > Decimal::ZERO { slot.1 += 1; } else { slot.2 += 1; }
            slot.3 += net;
            slot.4 += fees;
            slot.5 += f1 + f2;
        }

        // Session metadata — driven by `FROM sessions` so sessions with zero
        // trades still appear in the output (matches the old LEFT JOIN shape).
        let mut stmt = self.conn.prepare(
            "SELECT s.id, s.started_at_ms, s.ended_at_ms, s.state, s.end_reason,
                    br.leverage, br.min_entry_spread_bps, br.exit_mode, br.max_order_size_usd
             FROM sessions s
             LEFT JOIN bot_runs br ON br.id = (
                 SELECT MIN(br2.id) FROM bot_runs br2 WHERE br2.session_id = s.id
             )
             ORDER BY s.started_at_ms DESC"
        )?;
        let rows: Vec<SessionSummaryRow> = stmt.query_map([], |row| {
            let id: i64 = row.get(0)?;
            let (trade_count, wins, losses, total_pnl, total_fees, total_funding) =
                agg.get(&id).cloned().unwrap_or((0, 0, 0, Decimal::ZERO, Decimal::ZERO, Decimal::ZERO));
            Ok(SessionSummaryRow {
                id,
                started_at_ms: row.get(1)?,
                ended_at_ms: row.get(2)?,
                state: row.get(3)?,
                end_reason: row.get(4)?,
                trade_count,
                wins,
                losses,
                total_pnl: dec_to_f64(total_pnl),
                total_fees: dec_to_f64(total_fees),
                total_funding: dec_to_f64(total_funding),
                leverage: row.get(5)?,
                min_entry_spread_bps: row.get(6)?,
                exit_mode: row.get(7)?,
                max_order_size_usd: row.get(8)?,
            })
        })?.filter_map(|r| r.ok()).collect();
        Ok(rows)
    }

    /// Load all-time aggregated stats across all sessions.
    pub fn load_all_time_stats(&self) -> Result<AllTimeStats> {
        let (session_count, total_uptime_ms): (i64, i64) = self.conn.query_row(
            "SELECT COUNT(*),
                    COALESCE(SUM(CASE
                        WHEN ended_at_ms IS NOT NULL THEN ended_at_ms - started_at_ms
                        ELSE CAST(strftime('%s','now') AS INTEGER) * 1000 - started_at_ms
                    END), 0)
             FROM sessions",
            [], |r| Ok((r.get(0)?, r.get(1)?)),
        )?;

        // All-time aggregates + per-session totals folded through `derive_pnl`
        // in Rust. Single source of truth: there is no SQL path that
        // independently computes gross/net PnL.
        use std::collections::HashMap;
        use rust_decimal::Decimal;

        let rows = self.load_trades_for_aggregation(None)?;
        let mut trade_count: i64 = 0;
        let mut wins: i64 = 0;
        let mut losses: i64 = 0;
        let mut total_pnl = Decimal::ZERO;
        let mut total_fees = Decimal::ZERO;
        let mut total_funding = Decimal::ZERO;
        // Per-session net PnL for best/worst resolution (replaces the
        // queries that referenced the dropped `trades.net_pnl` column).
        let mut per_session: HashMap<i64, Decimal> = HashMap::new();
        for t in rows {
            let (_gross, net, _pct) = derive_pnl(
                t.is_long_1, &t.qty,
                &t.entry_1, &t.entry_2, &t.exit_1, &t.exit_2,
                &t.fees, &t.funding_1, &t.funding_2,
            );
            let fees = Decimal::from_str(t.fees.trim()).unwrap_or(Decimal::ZERO);
            let f1 = Decimal::from_str(t.funding_1.trim()).unwrap_or(Decimal::ZERO);
            let f2 = Decimal::from_str(t.funding_2.trim()).unwrap_or(Decimal::ZERO);
            trade_count += 1;
            if net > Decimal::ZERO { wins += 1; } else { losses += 1; }
            total_pnl += net;
            total_fees += fees;
            total_funding += f1 + f2;
            *per_session.entry(t.session_id).or_insert(Decimal::ZERO) += net;
        }

        let (best_session_id, best_session_pnl) = per_session.iter()
            .max_by(|a, b| a.1.cmp(b.1))
            .map(|(k, v)| (Some(*k), dec_to_f64(*v)))
            .unwrap_or((None, 0.0));
        let (worst_session_id, worst_session_pnl) = per_session.iter()
            .min_by(|a, b| a.1.cmp(b.1))
            .map(|(k, v)| (Some(*k), dec_to_f64(*v)))
            .unwrap_or((None, 0.0));

        Ok(AllTimeStats {
            session_count,
            trade_count,
            wins,
            losses,
            total_pnl: dec_to_f64(total_pnl),
            total_fees: dec_to_f64(total_fees),
            total_funding: dec_to_f64(total_funding),
            total_uptime_ms,
            best_session_id,
            best_session_pnl,
            worst_session_id,
            worst_session_pnl,
        })
    }

    /// Load closed trades for a specific session (by session_id).
    pub fn load_closed_trades_for_session(&self, session_id: i64) -> Result<Vec<ClosedTradeRow>> {
        let mut stmt = self.conn.prepare(CLOSED_TRADES_SELECT_SESSION)?;
        let trades = stmt.query_map(params![session_id], |row| {
            Ok(closed_trade_row_from_sql(row))
        })?.filter_map(|r| r.ok()).collect();
        Ok(trades)
    }

    /// Load per-trade aggregation rows for session / all-time stats. Shared
    /// feeder for `load_all_sessions_summary` and `load_all_time_stats` —
    /// both fold `derive_pnl` over this and nothing else computes PnL.
    fn load_trades_for_aggregation(&self, session_filter: Option<i64>) -> Result<Vec<AggTradeRow>> {
        let mut stmt = self.conn.prepare(AGG_TRADES_SELECT)?;
        let rows = stmt.query_map(params![session_filter], |row| {
            Ok(AggTradeRow {
                session_id: row.get(0)?,
                is_long_1: row.get::<_, i64>(1).unwrap_or(1) != 0,
                qty: row.get(2)?,
                entry_1: row.get(3)?,
                entry_2: row.get(4)?,
                exit_1: row.get(5)?,
                exit_2: row.get(6)?,
                fees: row.get(7)?,
                funding_1: row.get(8)?,
                funding_2: row.get(9)?,
            })
        })?.filter_map(|r| r.ok()).collect();
        Ok(rows)
    }

    // ── Exchange health events ──────────────────────────────────────────────

    /// Insert an exchange health event (connect/disconnect/error/blocked).
    pub fn insert_exchange_event(
        &self,
        ts_ms: u64,
        exchange_id: i64,
        channel: &str,
        event: &str,
        detail: Option<&str>,
        session_dur_ms: Option<u64>,
    ) -> Result<()> {
        self.conn.execute(
            "INSERT INTO exchange_events (ts_ms, exchange_id, channel, event, detail, session_dur_ms)
             VALUES (?1, ?2, ?3, ?4, ?5, ?6)",
            params![ts_ms as i64, exchange_id, channel, event, detail, session_dur_ms.map(|v| v as i64)],
        )?;
        Ok(())
    }

    /// Update cached fee rates for an exchange.
    pub fn update_exchange_fees(&self, exchange_id: i64, fee_taker: &str, fee_maker: &str) -> Result<()> {
        self.conn.execute(
            "UPDATE exchanges SET fee_taker = ?1, fee_maker = ?2, fee_updated_at_ms = ?3 WHERE id = ?4",
            params![fee_taker, fee_maker, crate::utils::time::current_unix_ms() as i64, exchange_id],
        )?;
        Ok(())
    }

    /// Ensure an exchange exists in the DB. Returns its ID (existing or newly created).
    /// New exchanges get autoincrement IDs. Known exchanges (lighter=1, extended=2)
    /// are pre-seeded in Storage::open() with stable IDs.
    pub fn ensure_exchange(&self, name: &str) -> Result<i64> {
        // Try to find existing
        let existing: Option<i64> = self.conn.query_row(
            "SELECT id FROM exchanges WHERE name = ?1",
            params![name],
            |row| row.get(0),
        ).ok();
        if let Some(id) = existing {
            return Ok(id);
        }
        // Insert with autoincrement ID
        let now_ms = crate::utils::time::current_unix_ms();
        let display = format!("{}.exchange", name);
        self.conn.execute(
            "INSERT INTO exchanges (name, display_name, exchange_type, added_at_ms) VALUES (?1, ?2, 'unknown', ?3)",
            params![name, display, now_ms as i64],
        )?;
        Ok(self.conn.last_insert_rowid())
    }

    /// Load exchange name → DB ID mapping from exchanges table.
    pub fn load_exchange_ids(&self) -> Result<Vec<(String, i64)>> {
        let mut stmt = self.conn.prepare("SELECT name, id FROM exchanges ORDER BY id")?;
        let rows = stmt.query_map([], |row| {
            Ok((row.get::<_, String>(0)?, row.get::<_, i64>(1)?))
        })?;
        Ok(rows.filter_map(|r| r.ok()).collect())
    }

    /// Resolve a list of exchange names to their DB IDs in input order.
    /// Returns `Err` if any name doesn't exist in the `exchanges` table —
    /// callers should treat this as user error and abort the operation
    /// (rather than silently writing to a wrong row).
    pub fn resolve_exchange_ids(&self, names: &[&str]) -> Result<Vec<i64>> {
        let mut stmt = self.conn.prepare("SELECT id FROM exchanges WHERE name = ?1")?;
        let mut out = Vec::with_capacity(names.len());
        for name in names {
            let id: i64 = stmt
                .query_row(params![name], |r| r.get(0))
                .with_context(|| format!("exchange '{name}' not found in DB"))?;
            out.push(id);
        }
        Ok(out)
    }

    /// Symbols that are active on 2+ exchanges (candidates for arb overrides).
    pub fn load_active_common_symbols(&self) -> Result<Vec<String>> {
        let mut stmt = self.conn.prepare(
            "SELECT symbol FROM exchange_symbols
             GROUP BY symbol HAVING COUNT(DISTINCT exchange_id) >= 2
             ORDER BY symbol"
        )?;
        let rows = stmt.query_map([], |row| row.get::<_, String>(0))?;
        Ok(rows.filter_map(|r| r.ok()).collect())
    }

    /// (symbol, exchange_id) pairs for every listing — used by the UI
    /// to filter valid symbol↔exchange-pair combinations in override forms.
    pub fn load_active_symbol_exchanges(&self) -> Result<Vec<(String, i64)>> {
        let mut stmt = self.conn.prepare(
            "SELECT symbol, exchange_id FROM exchange_symbols
             ORDER BY symbol, exchange_id"
        )?;
        let rows = stmt.query_map([], |row| {
            Ok((row.get::<_, String>(0)?, row.get::<_, i64>(1)?))
        })?;
        Ok(rows.filter_map(|r| r.ok()).collect())
    }

    // ── Symbol blocks (persistent) ──────────────────────────────────────────

    /// Insert a new active block. Returns the row ID.
    pub fn insert_symbol_block(
        &self,
        symbol: &str,
        exchange_id: Option<i64>,
        exchange_id_2: Option<i64>,
        block_type: &str,
        reason: &str,
    ) -> Result<i64> {
        let now_ms = crate::utils::time::current_unix_ms() as i64;
        self.conn.execute(
            "INSERT INTO symbol_blocks (symbol, exchange_id, exchange_id_2, block_type, reason, blocked_at_ms)
             VALUES (?1, ?2, ?3, ?4, ?5, ?6)",
            params![symbol, exchange_id, exchange_id_2, block_type, reason, now_ms],
        )?;
        Ok(self.conn.last_insert_rowid())
    }

    /// Unblock: set unblocked_at_ms for active blocks matching symbol + block_type + exchange_id + exchange_id_2.
    /// exchange_id=None matches only rows with NULL exchange_id (global blocks).
    pub fn unblock_symbol(
        &self,
        symbol: &str,
        exchange_id: Option<i64>,
        exchange_id_2: Option<i64>,
        block_type: &str,
        unblock_reason: &str,
    ) -> Result<u64> {
        let now_ms = crate::utils::time::current_unix_ms() as i64;
        let sql = match (exchange_id, exchange_id_2) {
            (Some(eid1), Some(eid2)) => {
                return Ok(self.conn.execute(
                    "UPDATE symbol_blocks SET unblocked_at_ms = ?1, unblock_reason = ?2
                     WHERE symbol = ?3 AND block_type = ?4 AND exchange_id = ?5 AND exchange_id_2 = ?6 AND unblocked_at_ms IS NULL",
                    params![now_ms, unblock_reason, symbol, block_type, eid1, eid2],
                )? as u64);
            }
            (Some(eid), None) => {
                return Ok(self.conn.execute(
                    "UPDATE symbol_blocks SET unblocked_at_ms = ?1, unblock_reason = ?2
                     WHERE symbol = ?3 AND block_type = ?4 AND exchange_id = ?5 AND exchange_id_2 IS NULL AND unblocked_at_ms IS NULL",
                    params![now_ms, unblock_reason, symbol, block_type, eid],
                )? as u64);
            }
            _ => {
                "UPDATE symbol_blocks SET unblocked_at_ms = ?1, unblock_reason = ?2
                 WHERE symbol = ?3 AND block_type = ?4 AND exchange_id IS NULL AND unblocked_at_ms IS NULL"
            }
        };
        Ok(self.conn.execute(sql, params![now_ms, unblock_reason, symbol, block_type])? as u64)
    }

    /// Unblock a specific block by its row ID. Returns the block data for DashMap update.
    pub fn unblock_by_id(&self, id: i64, unblock_reason: &str) -> Result<Option<SymbolBlock>> {
        let now_ms = crate::utils::time::current_unix_ms() as i64;
        // Read block data before unblocking
        let mut stmt = self.conn.prepare(
            "SELECT symbol, exchange_id, exchange_id_2, block_type, reason, blocked_at_ms
             FROM symbol_blocks WHERE id = ?1 AND unblocked_at_ms IS NULL"
        )?;
        let mut rows = stmt.query_map(params![id], |row| {
            Ok(SymbolBlock {
                db_id: Some(id),
                symbol: row.get(0)?,
                exchange_id: row.get(1)?,
                exchange_id_2: row.get(2)?,
                block_type: BlockType::from_tag(row.get::<_, String>(3)?.as_str()),
                reason: row.get(4)?,
                blocked_at_ms: row.get::<_, i64>(5)? as u64,
                expires_at_ms: None,
            })
        })?;
        let block = rows.next().transpose()?;
        if block.is_some() {
            self.conn.execute(
                "UPDATE symbol_blocks SET unblocked_at_ms = ?1, unblock_reason = ?2 WHERE id = ?3",
                params![now_ms, unblock_reason, id],
            )?;
        }
        Ok(block)
    }

    /// Load all active blocks (unblocked_at_ms IS NULL).
    pub fn load_active_blocks(&self) -> Result<Vec<(String, SymbolBlock)>> {
        let mut stmt = self.conn.prepare(
            "SELECT id, symbol, exchange_id, exchange_id_2, block_type, reason, blocked_at_ms
             FROM symbol_blocks
             WHERE unblocked_at_ms IS NULL"
        )?;
        let rows = stmt.query_map([], |row| {
            let symbol: String = row.get(1)?;
            let block = SymbolBlock {
                db_id: Some(row.get(0)?),
                symbol: symbol.clone(),
                exchange_id: row.get(2)?,
                exchange_id_2: row.get(3)?,
                block_type: BlockType::from_tag(row.get::<_, String>(4)?.as_str()),
                reason: row.get(5)?,
                blocked_at_ms: row.get::<_, i64>(6)? as u64,
                expires_at_ms: None,
            };
            Ok((symbol, block))
        })?;
        rows.collect::<std::result::Result<Vec<_>, _>>()
            .context("Failed to load active symbol blocks")
    }

    // ── Trading halts ──────────────────────────────────────────────────────

    /// Insert a new trading halt record.
    ///
    /// `affected_exchanges` is `None` for global halts (kill switch, manual
    /// pause, permanent escalations) and `Some(csv)` for per-exchange halts
    /// where `csv` is comma-separated `ExchangeId::db_id()` values, e.g. `"1,3"`.
    pub fn insert_trading_halt(&self, halt_type: &str, detail: &str, halted_at_ms: u64, affected_exchanges: Option<&str>) -> Result<i64> {
        self.conn.execute(
            "INSERT INTO trading_halts (halt_type, detail, halted_at_ms, affected_exchanges) VALUES (?1, ?2, ?3, ?4)",
            params![halt_type, detail, halted_at_ms as i64, affected_exchanges],
        )?;
        Ok(self.conn.last_insert_rowid())
    }

    /// Close all active halts from previous session (called at startup).
    pub fn close_stale_trading_halts(&self) -> Result<u64> {
        let now_ms = crate::utils::time::current_unix_ms();
        let count = self.conn.execute(
            "UPDATE trading_halts SET resumed_at_ms = ?1, resume_reason = 'bot restarted'
             WHERE resumed_at_ms IS NULL",
            params![now_ms as i64],
        )?;
        Ok(count as u64)
    }

    /// Resume an active trading halt: set resumed_at_ms and resume_reason.
    /// Updates the most recent active (un-resumed) halt of the given type.
    pub fn resume_trading_halt(&self, halt_type: &str, resume_reason: &str, resumed_at_ms: u64) -> Result<u64> {
        let updated = self.conn.execute(
            "UPDATE trading_halts SET resumed_at_ms = ?1, resume_reason = ?2
             WHERE halt_type = ?3 AND resumed_at_ms IS NULL",
            params![resumed_at_ms as i64, resume_reason, halt_type],
        )?;
        Ok(updated as u64)
    }

    // ── Optimizer ───────────────────────────────────────────────────────────

    /// Insert a new optimizer run, return its ID.
    pub fn insert_optimizer_run(
        &self,
        symbol: Option<&str>,
        mode: &str,
        config_json: &str,
        started_at_ms: i64,
    ) -> Result<i64> {
        self.conn.execute(
            "INSERT INTO optimizer_runs (symbol, mode, status, started_at_ms, config_json)
             VALUES (?1, ?2, 'running', ?3, ?4)",
            params![symbol, mode, started_at_ms, config_json],
        )?;
        Ok(self.conn.last_insert_rowid())
    }

    /// Mark an optimizer run as completed with summary stats.
    pub fn finish_optimizer_run(
        &self,
        run_id: i64,
        status: &str,
        finished_at_ms: i64,
        total_combos: usize,
        evaluated: usize,
        elapsed_secs: f64,
        best_sharpe: f64,
        best_profit: f64,
        tick_count: usize,
        tick_range_desc: &str,
        csv_path: &str,
    ) -> Result<()> {
        self.conn.execute(
            "UPDATE optimizer_runs SET status=?2, finished_at_ms=?3,
             total_combos=?4, evaluated=?5, elapsed_secs=?6,
             best_sharpe=?7, best_profit=?8, tick_count=?9,
             tick_range_desc=?10, csv_path=?11
             WHERE id=?1",
            params![run_id, status, finished_at_ms,
                    total_combos as i64, evaluated as i64, elapsed_secs,
                    best_sharpe, best_profit, tick_count as i64,
                    tick_range_desc, csv_path],
        )?;
        Ok(())
    }

    /// Insert a top-N result row for an optimizer run.
    #[allow(clippy::too_many_arguments)] // one optimizer-result row = 20+ independent metric/param columns
    pub fn insert_optimizer_result(
        &self,
        run_id: i64,
        rank: usize,
        entry_spread_bps: f64,
        take_profit_pct: f64,
        min_exit_spread_bps: f64,
        signal_cooldown_s: f64,
        min_exit_checks: u32,
        min_top_level_usd: f64,
        net_profit: f64,
        trade_count: usize,
        win_rate: f64,
        sharpe_ratio: f64,
        profit_factor: f64,
        max_drawdown: f64,
        avg_hold_secs: f64,
        tp_count: usize,
        tp_pnl: f64,
        sl_count: usize,
        sl_pnl: f64,
        abort_count: usize,
        abort_pnl: f64,
        entry_spike_bps: f64,
        exit_spike_bps: f64,
        max_pos_per_symbol: usize,
        peak_open_positions: usize,
        peak_used_capital: f64,
    ) -> Result<()> {
        self.conn.execute(
            "INSERT INTO optimizer_results (
                run_id, rank, entry_spread_bps, take_profit_pct, min_exit_spread_bps, signal_cooldown_s,
                min_exit_checks, min_top_level_usd,
                net_profit, trade_count, win_rate, sharpe_ratio, profit_factor,
                max_drawdown, avg_hold_secs, tp_count, tp_pnl, sl_count, sl_pnl,
                abort_count, abort_pnl, entry_spike_bps, exit_spike_bps, max_pos_per_symbol,
                peak_open_positions, peak_used_capital
            ) VALUES (?1,?2,?3,?4,?5,?6,?7,?8,?9,?10,?11,?12,?13,?14,?15,?16,?17,?18,?19,?20,?21,?22,?23,?24,?25,?26)",
            params![run_id, rank as i64, entry_spread_bps, take_profit_pct, min_exit_spread_bps, signal_cooldown_s,
                    min_exit_checks, min_top_level_usd,
                    net_profit, trade_count as i64, win_rate, sharpe_ratio, profit_factor,
                    max_drawdown, avg_hold_secs, tp_count as i64, tp_pnl,
                    sl_count as i64, sl_pnl, abort_count as i64, abort_pnl,
                    entry_spike_bps, exit_spike_bps, max_pos_per_symbol as i64,
                    peak_open_positions as i64, peak_used_capital],
        )?;
        Ok(())
    }

    /// List all optimizer runs (most recent first).
    pub fn list_optimizer_runs(&self) -> Result<Vec<serde_json::Value>> {
        let mut stmt = self.conn.prepare(
            "SELECT r.id, r.symbol, r.mode, r.status, r.started_at_ms, r.finished_at_ms,
                    r.total_combos, r.evaluated, r.elapsed_secs,
                    r.best_sharpe, r.best_profit, r.tick_count, r.tick_range_desc, r.csv_path,
                    r.config_json,
                    res.trade_count, res.peak_open_positions, res.peak_used_capital
             FROM optimizer_runs r
             LEFT JOIN optimizer_results res ON res.run_id = r.id AND res.rank = 1
             ORDER BY r.started_at_ms DESC"
        )?;
        let rows = stmt.query_map([], |row| {
            Ok(serde_json::json!({
                "id": row.get::<_, i64>(0)?,
                "symbol": row.get::<_, Option<String>>(1)?,
                "mode": row.get::<_, String>(2)?,
                "status": row.get::<_, String>(3)?,
                "started_at_ms": row.get::<_, i64>(4)?,
                "finished_at_ms": row.get::<_, Option<i64>>(5)?,
                "total_combos": row.get::<_, Option<i64>>(6)?,
                "evaluated": row.get::<_, Option<i64>>(7)?,
                "elapsed_secs": row.get::<_, Option<f64>>(8)?,
                "best_sharpe": row.get::<_, Option<f64>>(9)?,
                "best_profit": row.get::<_, Option<f64>>(10)?,
                "tick_count": row.get::<_, Option<i64>>(11)?,
                "tick_range_desc": row.get::<_, Option<String>>(12)?,
                "csv_path": row.get::<_, Option<String>>(13)?,
                "config_json": row.get::<_, Option<String>>(14)?,
                "best_trades": row.get::<_, Option<i64>>(15)?,
                "best_peak_pos": row.get::<_, Option<i64>>(16)?,
                "best_peak_capital": row.get::<_, Option<f64>>(17)?,
            }))
        })?;
        rows.collect::<std::result::Result<Vec<_>, _>>()
            .context("Failed to list optimizer runs")
    }

    /// Get top-N results for a specific optimizer run.
    pub fn get_optimizer_results(&self, run_id: i64) -> Result<Vec<serde_json::Value>> {
        let mut stmt = self.conn.prepare(
            "SELECT rank, entry_spread_bps, take_profit_pct, min_exit_spread_bps, signal_cooldown_s,
                    min_exit_checks, min_top_level_usd,
                    net_profit, trade_count, win_rate, sharpe_ratio, profit_factor,
                    max_drawdown, avg_hold_secs,
                    tp_count, tp_pnl, sl_count, sl_pnl, abort_count, abort_pnl,
                    entry_spike_bps, exit_spike_bps, max_pos_per_symbol,
                    peak_open_positions, peak_used_capital
             FROM optimizer_results WHERE run_id = ?1 ORDER BY rank"
        )?;
        let rows = stmt.query_map(params![run_id], |row| {
            Ok(serde_json::json!({
                "rank": row.get::<_, i64>(0)?,
                "entry_spread_bps": row.get::<_, f64>(1)?,
                "take_profit_pct": row.get::<_, f64>(2)?,
                "min_exit_spread_bps": row.get::<_, Option<f64>>(3)?.unwrap_or(0.0),
                "signal_cooldown_s": row.get::<_, f64>(4)?,
                "min_exit_checks": row.get::<_, i64>(5)?,
                "min_top_level_usd": row.get::<_, f64>(6)?,
                "net_profit": row.get::<_, f64>(7)?,
                "trade_count": row.get::<_, i64>(8)?,
                "win_rate": row.get::<_, f64>(9)?,
                "sharpe_ratio": row.get::<_, f64>(10)?,
                "profit_factor": row.get::<_, f64>(11)?,
                "max_drawdown": row.get::<_, f64>(12)?,
                "avg_hold_secs": row.get::<_, f64>(13)?,
                "tp_count": row.get::<_, i64>(14)?,
                "tp_pnl": row.get::<_, f64>(15)?,
                "sl_count": row.get::<_, i64>(16)?,
                "sl_pnl": row.get::<_, f64>(17)?,
                "abort_count": row.get::<_, i64>(18)?,
                "abort_pnl": row.get::<_, f64>(19)?,
                "spike_bps": row.get::<_, Option<f64>>(20)?.unwrap_or(0.0),
                "max_pos_per_symbol": row.get::<_, Option<i64>>(22)?.unwrap_or(1),
                "peak_open_positions": row.get::<_, Option<i64>>(23)?.unwrap_or(0),
                "peak_used_capital": row.get::<_, Option<f64>>(24)?.unwrap_or(0.0),
            }))
        })?;
        rows.collect::<std::result::Result<Vec<_>, _>>()
            .context("Failed to get optimizer results")
    }

    // ── Symbol overrides ────────────────────────────────────────────────────

    pub fn load_symbol_overrides(&self) -> Result<Vec<SymbolOverrideRow>> {
        let mut stmt = self.conn.prepare(
            "SELECT symbol, exchange_id_1, exchange_id_2,
                    min_entry_spread_bps, take_profit_pct, min_exit_spread_bps,
                    exit_mode, signal_cooldown_secs, min_exit_checks,
                    min_top_level_usd, tick_max_spike_bps,
                    max_recv_staleness_secs, max_tick_age_ms,
                    max_positions_per_symbol, source_run_id, updated_at_ms
             FROM symbol_overrides ORDER BY symbol"
        )?;
        let rows = stmt.query_map([], |row| {
            Ok(SymbolOverrideRow {
                symbol: row.get(0)?,
                exchange_id_1: row.get(1)?,
                exchange_id_2: row.get(2)?,
                min_entry_spread_bps: row.get(3)?,
                take_profit_pct: row.get(4)?,
                min_exit_spread_bps: row.get(5)?,
                exit_mode: row.get(6)?,
                signal_cooldown_secs: row.get(7)?,
                min_exit_checks: row.get(8)?,
                min_top_level_usd: row.get(9)?,
                tick_max_spike_bps: row.get(10)?,
                max_recv_staleness_secs: row.get(11)?,
                max_tick_age_ms: row.get(12)?,
                max_positions_per_symbol: row.get(13)?,
                source_run_id: row.get(14)?,
                updated_at_ms: row.get(15)?,
            })
        })?;
        rows.collect::<std::result::Result<Vec<_>, _>>()
            .context("Failed to load symbol overrides")
    }

    pub fn upsert_symbol_override(&self, r: &SymbolOverrideRow) -> Result<()> {
        let (id1, id2) = if r.exchange_id_1 < r.exchange_id_2 {
            (r.exchange_id_1, r.exchange_id_2)
        } else {
            (r.exchange_id_2, r.exchange_id_1)
        };
        self.conn.execute(
            "INSERT INTO symbol_overrides (
                symbol, exchange_id_1, exchange_id_2,
                min_entry_spread_bps, take_profit_pct, min_exit_spread_bps,
                exit_mode, signal_cooldown_secs, min_exit_checks,
                min_top_level_usd, tick_max_spike_bps,
                max_recv_staleness_secs, max_tick_age_ms,
                max_positions_per_symbol, source_run_id, updated_at_ms
            ) VALUES (?1,?2,?3,?4,?5,?6,?7,?8,?9,?10,?11,?12,?13,?14,?15,?16)
            ON CONFLICT(symbol, exchange_id_1, exchange_id_2) DO UPDATE SET
                min_entry_spread_bps=?4, take_profit_pct=?5, min_exit_spread_bps=?6,
                exit_mode=?7, signal_cooldown_secs=?8, min_exit_checks=?9,
                min_top_level_usd=?10, tick_max_spike_bps=?11,
                max_recv_staleness_secs=?12, max_tick_age_ms=?13,
                max_positions_per_symbol=?14,
                source_run_id=?15, updated_at_ms=?16",
            params![
                r.symbol, id1, id2,
                r.min_entry_spread_bps, r.take_profit_pct, r.min_exit_spread_bps,
                r.exit_mode, r.signal_cooldown_secs, r.min_exit_checks,
                r.min_top_level_usd, r.tick_max_spike_bps,
                r.max_recv_staleness_secs, r.max_tick_age_ms,
                r.max_positions_per_symbol,
                r.source_run_id, r.updated_at_ms,
            ],
        )?;
        Ok(())
    }

    pub fn delete_symbol_override(&self, symbol: &str, ex_id_1: i64, ex_id_2: i64) -> Result<()> {
        let (id1, id2) = if ex_id_1 < ex_id_2 { (ex_id_1, ex_id_2) } else { (ex_id_2, ex_id_1) };
        self.conn.execute(
            "DELETE FROM symbol_overrides WHERE symbol = ?1 AND exchange_id_1 = ?2 AND exchange_id_2 = ?3",
            params![symbol, id1, id2],
        )?;
        Ok(())
    }
}

#[derive(Debug, Clone, serde::Serialize)]
pub struct SymbolOverrideRow {
    pub symbol: String,
    pub exchange_id_1: i64,
    pub exchange_id_2: i64,
    pub min_entry_spread_bps: Option<f64>,
    pub take_profit_pct: Option<f64>,
    pub min_exit_spread_bps: Option<f64>,
    pub exit_mode: Option<String>,
    pub signal_cooldown_secs: Option<f64>,
    pub min_exit_checks: Option<i32>,
    pub min_top_level_usd: Option<f64>,
    pub tick_max_spike_bps: Option<f64>,
    pub max_recv_staleness_secs: Option<f64>,
    pub max_tick_age_ms: Option<f64>,
    pub max_positions_per_symbol: Option<i32>,
    pub source_run_id: Option<i64>,
    pub updated_at_ms: i64,
}

// ── Helpers ──────────────────────────────────────────────────────────────────

fn parse_decimal(s: &str) -> Decimal {
    Decimal::from_str(s).unwrap_or(Decimal::ZERO)
}


fn parse_state(s: &str) -> PositionState {
    match s {
        "closing" => PositionState::Closing,
        "filling" => PositionState::Filling,
        _ => PositionState::Active,
    }
}

/// Shared SELECT for closed trades — used by `load_session_closed_trades` and
/// `load_closed_trades_for_session`. Both query paths differ only in the
/// WHERE clause on `session_id`; everything else — JOINs, subqueries, column
/// order — is identical, so pulling it into one constant + one row-builder
/// function ensures they can never drift.
///
/// Columns (in order) — see `closed_trade_row_from_sql` for the indices:
///   0  t.id
///   1  t.symbol
///   2  t.is_long_1
///   3  t.exchange_id_1
///   4  t.exchange_id_2
///   5  t.qty
///   6  t.state
///   7  t.opened_at_ms
///   8  t.closed_at_ms
///   9  fees            (derived: SUM(orders.fee) subquery)
///   10 t.exit_reason
///   11 session_seq
///   12 t.signal_at_ms
///   13 entry_price_1   (oe1.avg_fill_price)
///   14 entry_price_2   (oe2.avg_fill_price)
///   15 exit_price_1    (latest filled exit order on exchange_1)
///   16 exit_price_2    (latest filled exit order on exchange_2)
///   17 t.entry_spread_bps
///   18-24 cfg_* (7 cols for build_cfg_from_row)
///   25 t.funding_1
///   26 t.funding_2
const CLOSED_TRADES_SELECT_SESSION: &str = "
    SELECT t.id, t.symbol, t.is_long_1, t.exchange_id_1, t.exchange_id_2,
           t.qty, t.state, t.opened_at_ms, t.closed_at_ms,
           COALESCE((SELECT printf('%.18f', SUM(CAST(fee AS REAL)))
                     FROM orders WHERE trade_id = t.id), '0'),
           t.exit_reason,
           COALESCE(t.session_seq, 0),
           t.signal_at_ms,
           COALESCE(oe1.avg_fill_price, '0'), COALESCE(oe2.avg_fill_price, '0'),
           COALESCE((SELECT avg_fill_price FROM orders
                     WHERE trade_id = t.id AND exchange_id = t.exchange_id_1
                       AND purpose LIKE 'exit_%' AND status = 'filled'
                     ORDER BY first_fill_at_ms DESC, order_id DESC LIMIT 1), '0'),
           COALESCE((SELECT avg_fill_price FROM orders
                     WHERE trade_id = t.id AND exchange_id = t.exchange_id_2
                       AND purpose LIKE 'exit_%' AND status = 'filled'
                     ORDER BY first_fill_at_ms DESC, order_id DESC LIMIT 1), '0'),
           COALESCE(t.entry_spread_bps, '0'),
           t.cfg_entry_bps, t.cfg_include_fees_in_entry,
           t.cfg_exit_mode, t.cfg_exit_bps, t.cfg_exit_pnl_pct, t.cfg_include_fees_in_exit,
           t.cfg_leverage,
           COALESCE(t.funding_1, '0'), COALESCE(t.funding_2, '0')
    FROM trades t
    LEFT JOIN orders oe1 ON oe1.trade_id = t.id AND oe1.exchange_id = t.exchange_id_1 AND oe1.purpose LIKE 'entry_%'
    LEFT JOIN orders oe2 ON oe2.trade_id = t.id AND oe2.exchange_id = t.exchange_id_2 AND oe2.purpose LIKE 'entry_%'
    WHERE t.state IN ('closed', 'failed_open', 'closed_error') AND t.session_id = ?1
    ORDER BY t.closed_at_ms DESC
";

/// Build a `ClosedTradeRow` from a row produced by `CLOSED_TRADES_SELECT_SESSION`.
/// PnL (gross/net/pct) is derived here from the per-order data and funding —
/// no denormalized columns on `trades`.
fn closed_trade_row_from_sql(row: &rusqlite::Row) -> ClosedTradeRow {
    let qty: String = row.get(5).unwrap_or_else(|_| "0".into());
    let is_long_1 = row.get::<_, i64>(2).unwrap_or(1) != 0;
    let fees: String = row.get(9).unwrap_or_else(|_| "0".into());
    let entry_price_1: String = row.get(13).unwrap_or_else(|_| "0".into());
    let entry_price_2: String = row.get(14).unwrap_or_else(|_| "0".into());
    let exit_price_1: String = row.get(15).unwrap_or_else(|_| "0".into());
    let exit_price_2: String = row.get(16).unwrap_or_else(|_| "0".into());
    let funding_1: String = row.get(25).unwrap_or_else(|_| "0".into());
    let funding_2: String = row.get(26).unwrap_or_else(|_| "0".into());

    let (gross, net, net_pct) = derive_pnl(
        is_long_1, &qty,
        &entry_price_1, &entry_price_2,
        &exit_price_1, &exit_price_2,
        &fees,
        &funding_1, &funding_2,
    );

    ClosedTradeRow {
        id: row.get(0).unwrap_or_default(),
        symbol: row.get(1).unwrap_or_default(),
        is_long_1,
        exchange_id_1: row.get(3).unwrap_or(0),
        exchange_id_2: row.get(4).unwrap_or(0),
        qty,
        state: row.get(6).unwrap_or_default(),
        opened_at_ms: row.get(7).unwrap_or(0),
        closed_at_ms: row.get(8).unwrap_or(0),
        gross_pnl: dec(gross),
        fees,
        net_pnl: dec(net),
        net_pnl_pct: dec(net_pct),
        exit_reason: row.get(10).unwrap_or_default(),
        session_seq: row.get::<_, i64>(11).unwrap_or(0),
        signal_at_ms: row.get::<_, u64>(12).unwrap_or(0),
        entry_price_1,
        entry_price_2,
        exit_price_1,
        exit_price_2,
        entry_spread_bps: row.get::<_, String>(17).unwrap_or_default(),
        cfg: build_cfg_from_row(row, 18),
        funding_1,
        funding_2,
    }
}

/// Lightweight per-trade row used by session / all-time aggregates. Fed
/// directly into `derive_pnl` so the aggregate path produces bit-exact
/// identical numbers to `load_closed_trades_for_session` — there is only
/// one place that computes gross/net PnL.
struct AggTradeRow {
    session_id: i64,
    is_long_1: bool,
    qty: String,
    entry_1: String,
    entry_2: String,
    exit_1: String,
    exit_2: String,
    fees: String,
    funding_1: String,
    funding_2: String,
}

/// SELECT for `load_trades_for_aggregation`. Fees use the same
/// `printf('%.18f', SUM(...))` formatting as `CLOSED_TRADES_SELECT_SESSION`
/// so the string passed to `derive_pnl` is identical on both paths.
/// Parameter `?1` is an optional session filter (NULL = all sessions).
const AGG_TRADES_SELECT: &str = "
    SELECT t.session_id, t.is_long_1, t.qty,
           COALESCE(oe1.avg_fill_price, '0'),
           COALESCE(oe2.avg_fill_price, '0'),
           COALESCE((SELECT avg_fill_price FROM orders
                     WHERE trade_id = t.id AND exchange_id = t.exchange_id_1
                       AND purpose LIKE 'exit_%' AND status = 'filled'
                     ORDER BY first_fill_at_ms DESC, order_id DESC LIMIT 1), '0'),
           COALESCE((SELECT avg_fill_price FROM orders
                     WHERE trade_id = t.id AND exchange_id = t.exchange_id_2
                       AND purpose LIKE 'exit_%' AND status = 'filled'
                     ORDER BY first_fill_at_ms DESC, order_id DESC LIMIT 1), '0'),
           COALESCE((SELECT printf('%.18f', SUM(CAST(fee AS REAL)))
                     FROM orders WHERE trade_id = t.id), '0'),
           COALESCE(t.funding_1, '0'),
           COALESCE(t.funding_2, '0')
    FROM trades t
    LEFT JOIN orders oe1 ON oe1.trade_id = t.id AND oe1.exchange_id = t.exchange_id_1 AND oe1.purpose LIKE 'entry_%'
    LEFT JOIN orders oe2 ON oe2.trade_id = t.id AND oe2.exchange_id = t.exchange_id_2 AND oe2.purpose LIKE 'entry_%'
    WHERE t.state IN ('closed','failed_open','closed_error')
      AND t.session_id IS NOT NULL
      AND (?1 IS NULL OR t.session_id = ?1)
";

/// Convert a `Decimal` to `f64` for JSON-facing struct fields
/// (`SessionSummaryRow.total_pnl`, `AllTimeStats.total_pnl`, etc.).
fn dec_to_f64(d: rust_decimal::Decimal) -> f64 {
    use rust_decimal::prelude::ToPrimitive;
    d.to_f64().unwrap_or(0.0)
}

/// Build StrategySnapshot from 7 consecutive columns starting at `offset`.
/// Returns None if any required field is NULL (legacy trades predating cfg_* columns).
fn build_cfg_from_row(row: &rusqlite::Row, offset: usize) -> Option<StrategySnapshot> {
    let entry_bps: Option<String> = row.get(offset).ok();
    let include_fees_in_entry: Option<i64> = row.get(offset + 1).ok();
    let exit_mode: Option<String> = row.get(offset + 2).ok();
    let exit_bps: Option<String> = row.get(offset + 3).ok();
    let exit_pnl_pct: Option<String> = row.get(offset + 4).ok();
    let include_fees_in_exit: Option<i64> = row.get(offset + 5).ok();
    let leverage: Option<i64> = row.get(offset + 6).ok();
    Some(StrategySnapshot {
        entry_bps: parse_decimal(&entry_bps?),
        include_fees_in_entry: include_fees_in_entry? != 0,
        exit_mode: exit_mode?,
        exit_bps: parse_decimal(&exit_bps?),
        exit_pnl_pct: parse_decimal(&exit_pnl_pct?),
        include_fees_in_exit: include_fees_in_exit? != 0,
        leverage: leverage? as u16,
    })
}

// ── Schema ───────────────────────────────────────────────────────────────────

const SCHEMA: &str = "
CREATE TABLE IF NOT EXISTS exchange_symbols (
    symbol       TEXT NOT NULL,
    exchange_id  INTEGER NOT NULL REFERENCES exchanges(id),
    market_name     TEXT NOT NULL DEFAULT '',
    size_decimals   INTEGER NOT NULL DEFAULT 0,
    price_decimals  INTEGER NOT NULL DEFAULT 0,
    qty_step        TEXT NOT NULL DEFAULT '0',
    min_qty         TEXT NOT NULL DEFAULT '0',
    max_leverage    INTEGER NOT NULL DEFAULT 20,
    mmf             REAL,
    limit_price_cap REAL,
    UNIQUE(symbol, exchange_id)
);
CREATE INDEX IF NOT EXISTS idx_exchange_symbols ON exchange_symbols(symbol);

CREATE TABLE IF NOT EXISTS exchange_symbol_overrides (
    symbol       TEXT NOT NULL,
    exchange_id  INTEGER NOT NULL REFERENCES exchanges(id),
    field_name   TEXT NOT NULL,
    value        TEXT NOT NULL,
    updated_at   TEXT NOT NULL DEFAULT (datetime('now')),
    UNIQUE(symbol, exchange_id, field_name)
);

CREATE TABLE IF NOT EXISTS trades (
    id               TEXT PRIMARY KEY,
    session_id       INTEGER REFERENCES sessions(id),
    session_seq      INTEGER,
    symbol           TEXT NOT NULL,
    is_long_1        INTEGER NOT NULL DEFAULT 1,
    exchange_id_1    INTEGER NOT NULL DEFAULT 1 REFERENCES exchanges(id),
    exchange_id_2    INTEGER NOT NULL DEFAULT 2 REFERENCES exchanges(id),
    qty              TEXT NOT NULL,
    state            TEXT NOT NULL,
    opened_at_ms     INTEGER DEFAULT 0,
    closed_at_ms     INTEGER DEFAULT 0,
    -- NOTE: no `net_pnl`, `net_pnl_pct`, or `fees` columns. All three are
    -- derivatives of per-order data in `orders`:
    --   gross = f(entry/exit avg_fill_price, qty, is_long_1)
    --   fees  = SUM(orders.fee WHERE trade_id = t.id)
    --   net   = gross - fees + (funding_1 + funding_2)
    -- Only `funding_1`/`funding_2` are genuinely trade-level (accrued on the
    -- open position, not per-order) and stay on this table.
    exit_reason      TEXT DEFAULT '',
    -- signal_at_ms / exit_signal_at_ms are TRADE-LEVEL, not order-level, even
    -- though the dashboard uses them per-order to compute Recv→Sig latency.
    -- Reason: a signal is a single spread-detection event between two venues
    -- that fires both legs of the trade simultaneously — by construction the
    -- value is identical for entry_1 and entry_2 (and again for exit_1/exit_2).
    -- Storing it on the order would duplicate the same number twice per leg.
    -- Per-order resolution at read time happens via SQL CASE on `purpose`
    -- (see query_orders_for_trade / load_closed_orders_for_session).
    -- The in-memory equivalents are `OpenSignal.ts_ns` and `CloseSignal.ts_ns`
    -- in [bot/src/types.rs] — also single u64 per signal, not per leg.
    signal_at_ms             INTEGER DEFAULT 0,
    exit_signal_at_ms        INTEGER DEFAULT 0,
    entry_spread_bps  TEXT DEFAULT '0',
    -- Strategy config snapshot: thresholds/flags that drove the bot's decisions.
    -- entry_* frozen at trade open; exit_* updated on hot-reload and at close.
    cfg_entry_bps              TEXT,
    cfg_include_fees_in_entry  INTEGER,
    cfg_exit_mode              TEXT,
    cfg_exit_bps               TEXT,
    cfg_exit_pnl_pct           TEXT,
    cfg_include_fees_in_exit   INTEGER,
    cfg_leverage               INTEGER
);
CREATE INDEX IF NOT EXISTS idx_trades_state ON trades(state);

CREATE TABLE IF NOT EXISTS orders (
    order_id         INTEGER PRIMARY KEY,
    trade_id         TEXT NOT NULL REFERENCES trades(id),
    session_seq      INTEGER NOT NULL DEFAULT 0,
    symbol           TEXT NOT NULL,
    exchange_id      INTEGER NOT NULL REFERENCES exchanges(id),
    side             TEXT NOT NULL,
    order_type       TEXT NOT NULL,
    purpose          TEXT NOT NULL,
    qty              TEXT NOT NULL,
    limit_price      TEXT NOT NULL,
    status           TEXT NOT NULL DEFAULT 'placed',
    placed_at_ms     INTEGER NOT NULL,
    exchange_ts_ms   INTEGER DEFAULT 0,
    recv_ts_ms       INTEGER DEFAULT 0,
    signal_price     TEXT DEFAULT '0',
    filled_qty       TEXT DEFAULT '0',
    avg_fill_price   TEXT DEFAULT '0',
    first_fill_at_ms INTEGER DEFAULT 0,
    last_fill_at_ms  INTEGER DEFAULT 0,
    fee              TEXT DEFAULT '0',
    closed_at_ms     INTEGER DEFAULT 0,
    close_reason     TEXT DEFAULT '',
    tx_hash          TEXT,
    exchange_order_id TEXT
);
CREATE INDEX IF NOT EXISTS idx_orders_trade ON orders(trade_id);
CREATE INDEX IF NOT EXISTS idx_orders_status ON orders(status);

CREATE TABLE IF NOT EXISTS exchanges (
    id                INTEGER PRIMARY KEY,
    name              TEXT NOT NULL UNIQUE,
    display_name      TEXT NOT NULL,
    exchange_type     TEXT NOT NULL,
    fee_taker         TEXT,
    fee_maker         TEXT,
    fee_updated_at_ms INTEGER,
    added_at_ms       INTEGER NOT NULL
);

CREATE TABLE IF NOT EXISTS exchange_events (
    id             INTEGER PRIMARY KEY AUTOINCREMENT,
    ts_ms          INTEGER NOT NULL,
    exchange_id    INTEGER NOT NULL REFERENCES exchanges(id),
    channel        TEXT NOT NULL,
    event          TEXT NOT NULL,
    detail         TEXT,
    session_dur_ms INTEGER
);
CREATE INDEX IF NOT EXISTS idx_exchange_events_ts ON exchange_events(ts_ms);
CREATE INDEX IF NOT EXISTS idx_exchange_events_exchange ON exchange_events(exchange_id, ts_ms);

CREATE TABLE IF NOT EXISTS sessions (
    id              INTEGER PRIMARY KEY AUTOINCREMENT,
    started_at_ms   INTEGER NOT NULL,
    ended_at_ms     INTEGER,
    state           TEXT NOT NULL DEFAULT 'active',
    end_reason      TEXT
);
CREATE INDEX IF NOT EXISTS idx_sessions_active ON sessions(state) WHERE state = 'active';

-- Per-exchange equity snapshot captured at session start. Used by the UI
-- to display Balance Δ = Σ(current equity) − Σ(snapshot) as a sanity check
-- against Σ trade.pnl. If the two disagree, something accounting-wise is
-- not being recorded (e.g. 2026-04-18: HL naked shorts covered during
-- MEGA spike were never persisted as trades, UI said +$1.27 while the
-- real balance was −$10.29).
CREATE TABLE IF NOT EXISTS session_start_balances (
    session_id       INTEGER NOT NULL REFERENCES sessions(id),
    exchange_id      INTEGER NOT NULL REFERENCES exchanges(id),
    balance_usd      TEXT NOT NULL,
    captured_at_ms   INTEGER NOT NULL,
    PRIMARY KEY (session_id, exchange_id)
);

CREATE TABLE IF NOT EXISTS bot_runs (
    id                      INTEGER PRIMARY KEY AUTOINCREMENT,
    session_id              INTEGER REFERENCES sessions(id),
    started_at              TEXT NOT NULL,
    started_at_ms           INTEGER NOT NULL,
    min_entry_spread_bps    REAL NOT NULL,
    min_exit_pnl_pct        REAL NOT NULL,
    exit_mode               TEXT NOT NULL,
    leverage                INTEGER NOT NULL,
    max_order_size_usd      REAL NOT NULL,
    max_open_positions      INTEGER NOT NULL,
    max_positions_per_symbol INTEGER NOT NULL,
    max_used_capital_usd    REAL NOT NULL,
    stop_loss_buffer_pct    REAL NOT NULL,
    signal_cooldown_secs    REAL NOT NULL,
    blocklist               TEXT NOT NULL,
    state                   TEXT NOT NULL DEFAULT 'starting',
    running_at              TEXT,
    running_at_ms           INTEGER,
    stopped_at              TEXT,
    stopped_at_ms           INTEGER
);

CREATE TABLE IF NOT EXISTS symbol_blocks (
    id              INTEGER PRIMARY KEY AUTOINCREMENT,
    symbol          TEXT NOT NULL,
    exchange_id     INTEGER REFERENCES exchanges(id),
    block_type      TEXT NOT NULL,
    reason          TEXT NOT NULL,
    blocked_at_ms   INTEGER NOT NULL,
    unblocked_at_ms INTEGER,
    unblock_reason  TEXT
);
CREATE INDEX IF NOT EXISTS idx_symbol_blocks_active ON symbol_blocks(symbol) WHERE unblocked_at_ms IS NULL;

CREATE TABLE IF NOT EXISTS trading_halts (
    id              INTEGER PRIMARY KEY AUTOINCREMENT,
    halt_type       TEXT NOT NULL,
    detail          TEXT NOT NULL,
    halted_at_ms    INTEGER NOT NULL,
    resumed_at_ms   INTEGER,
    resume_reason   TEXT
);
CREATE INDEX IF NOT EXISTS idx_trading_halts_active ON trading_halts(halt_type) WHERE resumed_at_ms IS NULL;

CREATE TABLE IF NOT EXISTS optimizer_runs (
    id              INTEGER PRIMARY KEY AUTOINCREMENT,
    symbol          TEXT,
    mode            TEXT NOT NULL,
    status          TEXT NOT NULL DEFAULT 'running',
    started_at_ms   INTEGER NOT NULL,
    finished_at_ms  INTEGER,
    config_json     TEXT NOT NULL,
    total_combos    INTEGER,
    evaluated       INTEGER,
    elapsed_secs    REAL,
    best_sharpe     REAL,
    best_profit     REAL,
    tick_count      INTEGER,
    tick_range_desc TEXT,
    csv_path        TEXT
);
CREATE INDEX IF NOT EXISTS idx_opt_runs_status ON optimizer_runs(status);
CREATE INDEX IF NOT EXISTS idx_opt_runs_symbol ON optimizer_runs(symbol);

CREATE TABLE IF NOT EXISTS optimizer_results (
    id                  INTEGER PRIMARY KEY AUTOINCREMENT,
    run_id              INTEGER NOT NULL REFERENCES optimizer_runs(id),
    rank                INTEGER NOT NULL,
    entry_spread_bps    REAL,
    take_profit_pct     REAL,
    min_exit_spread_bps REAL,
    signal_cooldown_s   REAL,
    min_exit_checks     INTEGER,
    min_top_level_usd   REAL,
    net_profit          REAL,
    trade_count         INTEGER,
    win_rate            REAL,
    sharpe_ratio        REAL,
    profit_factor       REAL,
    max_drawdown        REAL,
    avg_hold_secs       REAL,
    tp_count            INTEGER,
    tp_pnl              REAL,
    sl_count            INTEGER,
    sl_pnl              REAL,
    abort_count         INTEGER,
    abort_pnl           REAL,
    entry_spike_bps     REAL,
    exit_spike_bps      REAL,
    max_pos_per_symbol  INTEGER,
    peak_open_positions INTEGER,
    peak_used_capital   REAL
);
CREATE INDEX IF NOT EXISTS idx_opt_results_run ON optimizer_results(run_id);

CREATE TABLE IF NOT EXISTS symbol_overrides (
    symbol          TEXT NOT NULL,
    exchange_id_1   INTEGER NOT NULL,
    exchange_id_2   INTEGER NOT NULL,
    min_entry_spread_bps     REAL,
    take_profit_pct          REAL,
    min_exit_spread_bps      REAL,
    exit_mode                TEXT,
    signal_cooldown_secs     REAL,
    min_exit_checks          INTEGER,
    min_top_level_usd        REAL,
    tick_max_spike_bps       REAL,
    max_recv_staleness_secs  REAL,
    max_tick_age_ms          REAL,
    max_positions_per_symbol INTEGER,
    source_run_id   INTEGER,
    updated_at_ms   INTEGER NOT NULL,
    PRIMARY KEY (symbol, exchange_id_1, exchange_id_2)
);

-- Signal audit log: records every entry/exit decision the bot acted on.
-- For entry signals, `trade_id` is backfilled by execute_open after the
-- trade row is created. For exit signals, `trade_id` is required and set
-- at insert time. The `legs` JSON column carries per-leg orderbook state
-- (top-N bids/asks, VWAP, exchange_ts) so any signal can be replayed for
-- debugging without depending on the recorded tick stream.
CREATE TABLE IF NOT EXISTS signals (
    id              TEXT PRIMARY KEY,
    session_id      INTEGER REFERENCES sessions(id),
    ts_ms           INTEGER NOT NULL,
    signal_type     TEXT NOT NULL,           -- 'entry' | 'exit'
    symbol          TEXT NOT NULL,

    -- Decision math (denormalized for fast filtering)
    spread_bps      TEXT NOT NULL,
    staleness_ms    INTEGER NOT NULL,        -- |max(leg ts) - min(leg ts)|

    -- Entry-only fields (NULL on exit)
    threshold_bps   TEXT,
    fee_deduct_bps  TEXT,
    order_qty       TEXT,
    order_qty_usd   TEXT,

    -- Exit-only fields (NULL on entry)
    exit_reason     TEXT,
    entry_spread_bps TEXT,
    gross_pnl_bps   TEXT,
    net_pnl_pct     TEXT,
    held_for_ms     INTEGER,
    funding_at_exit TEXT,

    -- Per-leg orderbook snapshot as JSON array. One element per exchange
    -- involved. Forward-compat for N>2 legs without schema changes.
    legs            TEXT NOT NULL,

    -- Trade linkage. NULL when:
    --   - entry signal hasn't been backfilled by execute_open yet, OR
    --   - signal was rejected and never created a trade.
    -- Always set for accepted exit signals (the position being closed).
    trade_id        TEXT REFERENCES trades(id),

    -- Reject reason. NULL for v1 (we only log accepted signals). Reserved
    -- for the future rejected-signals feature so adding it requires no
    -- ALTER TABLE: simply start populating this column with values like
    -- 'halt' / 'risk' / 'spike' / etc. and the schema is ready.
    rejected_reason TEXT
);
CREATE INDEX IF NOT EXISTS idx_signals_ts       ON signals(ts_ms);
CREATE INDEX IF NOT EXISTS idx_signals_symbol   ON signals(symbol);
CREATE INDEX IF NOT EXISTS idx_signals_trade    ON signals(trade_id);
CREATE INDEX IF NOT EXISTS idx_signals_session  ON signals(session_id);
CREATE INDEX IF NOT EXISTS idx_signals_type     ON signals(signal_type);
CREATE INDEX IF NOT EXISTS idx_signals_rejected ON signals(rejected_reason);
";

// ── Tests ────────────────────────────────────────────────────────────────────

#[cfg(test)]
mod tests {
    use super::*;
    use rust_decimal_macros::dec as decm;

    /// Unit tests for `derive_pnl` — the single source of truth for PnL
    /// computation. Any refactor that breaks one of these assertions breaks
    /// every UI/telegram/stats consumer of closed-trade PnL, so drift is
    /// detected at compile-test time.
    mod derive_pnl {
        use super::*;

        #[test]
        fn long_leg_1_profitable() {
            // is_long_1=true: buy ex_1 @100, sell ex_2 @101. Exit: sell ex_1 @102, buy ex_2 @99.
            // gross = (102-100)*1 + (101-99)*1 = 2 + 2 = 4
            // fees = 0.5; funding = 0.1 + 0.05 = 0.15
            // net = 4 - 0.5 + 0.15 = 3.65
            // notional = 1 * (100+101)/2 = 100.5
            // pct = 3.65 / 100.5 * 100 = 3.6318...
            let (gross, net, pct) = derive_pnl(
                true, "1",
                "100", "101",
                "102", "99",
                "0.5",
                "0.1", "0.05",
            );
            assert_eq!(gross, decm!(4));
            assert_eq!(net, decm!(3.65));
            // Use precise division for comparison
            let expected_pct = decm!(3.65) / decm!(100.5) * decm!(100);
            assert_eq!(pct, expected_pct);
        }

        #[test]
        fn long_leg_2_profitable() {
            // is_long_1=false: sell ex_1 @100, buy ex_2 @99. Exit: buy ex_1 @98, sell ex_2 @102.
            // gross = (100-98)*1 + (102-99)*1 = 2 + 3 = 5
            let (gross, net, _) = derive_pnl(
                false, "1",
                "100", "99",
                "98", "102",
                "0", "0", "0",
            );
            assert_eq!(gross, decm!(5));
            assert_eq!(net, decm!(5));
        }

        #[test]
        fn fees_are_subtracted_funding_is_added() {
            // Positive funding is received → net goes up. Fees always subtract.
            let (_, net, _) = derive_pnl(
                true, "10",
                "100", "100",
                "101", "99",          // gross = (101-100)*10 + (100-99)*10 = 20
                "5",                  // fees
                "1.5", "0.7",         // funding_1 + funding_2 = 2.2
            );
            // net = 20 - 5 + 2.2 = 17.2
            assert_eq!(net, decm!(17.2));
        }

        #[test]
        fn partial_close_returns_zero_gross() {
            // Missing exit_2 → unhedged close → gross must be zero
            // (net still reflects fees+funding but gross is zero).
            let (gross, net, pct) = derive_pnl(
                true, "1",
                "100", "100",
                "102", "0",           // exit_2 = 0
                "0.2", "0", "0",
            );
            assert_eq!(gross, Decimal::ZERO);
            // With zero gross: net = -fees = -0.2
            assert_eq!(net, decm!(-0.2));
            // Notional zero → pct zero.
            assert_eq!(pct, Decimal::ZERO);
        }

        #[test]
        fn failed_open_returns_zero_gross() {
            // No fills at all → everything zero.
            let (gross, net, pct) = derive_pnl(
                true, "1",
                "0", "0", "0", "0",
                "0", "0", "0",
            );
            assert_eq!(gross, Decimal::ZERO);
            assert_eq!(net, Decimal::ZERO);
            assert_eq!(pct, Decimal::ZERO);
        }

        #[test]
        fn failed_open_with_one_leg_and_real_fee() {
            // Exact shape of the prod EIGEN bug: only entry_1 filled
            // (qty=91, avg=0.97796), entry_2 / exits zero, and the one
            // filled leg incurred a real fee. gross must be zero — the
            // old SQL CTE produced 0.97796 * 91 ≈ +88.99 here, poisoning
            // session totals.
            let (gross, net, pct) = derive_pnl(
                false, "91",
                "0.97796", "0",
                "0", "0",
                "0.003300615",
                "0", "0",
            );
            assert_eq!(gross, Decimal::ZERO);
            assert_eq!(net, decm!(-0.003300615));
            assert_eq!(pct, Decimal::ZERO);
        }

        #[test]
        fn loss_case() {
            // Price moved the wrong way on both legs + fees → net negative.
            // is_long_1=true: (99-100)*1 + (100-101)*1 = -1 + -1 = -2 gross
            let (gross, net, _) = derive_pnl(
                true, "1",
                "100", "100",
                "99", "101",
                "0.3", "0", "0",
            );
            assert_eq!(gross, decm!(-2));
            assert_eq!(net, decm!(-2.3));
        }

        #[test]
        fn realistic_apt_trade_from_prod() {
            // Real numbers from prod trade #3 (APT) after the ZERO1 fee fix —
            // the exact scenario that triggered this whole refactor.
            //   Extended (ex_1) entry=0.8608, exit=0.861, is_long_1=false (short)
            //   ZERO1 (ex_2) entry=0.8528, exit=0.8575336206896551724137931034
            //   qty=17.4
            // gross = (entry_1 - exit_1)*qty + (exit_2 - entry_2)*qty
            //       = (0.8608-0.861)*17.4 + (0.8575336...-0.8528)*17.4
            //       = -0.00348 + 0.08236...
            //       = 0.07888...
            // fees (real, post-fix): 0.003370032 + 0.005193552 + 0.003370815 + 0.005222379750
            //                       = 0.017156778750
            // funding = 0 (no funding in this trade)
            let (gross, net, pct) = derive_pnl(
                false, "17.4",
                "0.8608", "0.8528",
                "0.861",  "0.8575336206896551724137931034",
                "0.017156778750",
                "0", "0",
            );
            // Gross from the exact arithmetic the UI will show:
            let expected_gross =
                (decm!(0.8608) - decm!(0.861)) * decm!(17.4)
              + (decm!(0.8575336206896551724137931034) - decm!(0.8528)) * decm!(17.4);
            assert_eq!(gross, expected_gross);

            let expected_net = expected_gross - decm!(0.017156778750);
            assert_eq!(net, expected_net);

            let notional = decm!(17.4) * (decm!(0.8608) + decm!(0.8528)) / decm!(2);
            let expected_pct = expected_net / notional * decm!(100);
            assert_eq!(pct, expected_pct);
        }
    }

    /// End-to-end test against an in-memory SQLite: build a fixture trade +
    /// 4 orders, load via `load_closed_trades_for_session`, assert the
    /// derived gross/fees/net/pct match the closed-form expectation.
    ///
    /// This exercises the full path: schema → row builder → derive_pnl,
    /// including the SQL JOINs and correlated subqueries for exit prices
    /// and per-trade fee sum.
    #[test]
    fn load_closed_trades_derives_pnl_correctly() {
        let db = Storage::open(":memory:").expect("open in-memory");

        // Fixture: one active session, one closed trade with 4 orders.
        let now_ms: i64 = 1_700_000_000_000;
        db.conn.execute(
            "INSERT INTO sessions (id, started_at_ms, state) VALUES (1, ?1, 'active')",
            params![now_ms],
        ).unwrap();

        // Closed trade: ex_1=Lighter (long), ex_2=Extended (short), qty=1.
        // Funding: +0.05 on leg 1, -0.02 on leg 2 → net +0.03.
        db.conn.execute(
            "INSERT INTO trades (
                id, session_id, session_seq, symbol, is_long_1,
                exchange_id_1, exchange_id_2, qty, state,
                opened_at_ms, closed_at_ms, signal_at_ms,
                exit_reason, funding_1, funding_2
             ) VALUES ('t1', 1, 1, 'BTC', 1,
                       1, 2, '1', 'closed',
                       ?1, ?2, ?1,
                       'Tp', '0.05', '-0.02')",
            params![now_ms, now_ms + 60_000],
        ).unwrap();

        // 4 orders: entry_1 (Lighter, buy), entry_2 (Extended, sell),
        //           exit_1 (Lighter, sell),  exit_2 (Extended, buy).
        // Prices chosen so the arithmetic is trivial.
        let insert_order = |oid: u64, exchange_id: i64, side: &str, purpose: &str,
                            avg_price: &str, fee: &str, fill_ms: i64| {
            db.conn.execute(
                "INSERT INTO orders (
                    order_id, trade_id, session_seq, symbol, exchange_id, side,
                    order_type, purpose, qty, limit_price, status,
                    filled_qty, avg_fill_price, fee,
                    first_fill_at_ms, last_fill_at_ms, closed_at_ms, placed_at_ms
                ) VALUES (?1, 't1', 1, 'BTC', ?2, ?3, 'ioc', ?4,
                          '1', ?5, 'filled',
                          '1', ?5, ?6,
                          ?7, ?7, ?7, ?7)",
                params![oid, exchange_id, side, purpose, avg_price, fee, fill_ms],
            ).unwrap();
        };
        insert_order(1, 1, "buy",  "entry_1", "100.0", "0.10", now_ms);
        insert_order(2, 2, "sell", "entry_2", "101.0", "0.11", now_ms);
        insert_order(3, 1, "sell", "exit_1",  "102.0", "0.12", now_ms + 60_000);
        insert_order(4, 2, "buy",  "exit_2",  "99.0", "0.09", now_ms + 60_000);

        // Load via the public API — this is the whole point of the test.
        let trades = db.load_closed_trades_for_session(1).expect("load");
        assert_eq!(trades.len(), 1);
        let t = &trades[0];

        // Expected:
        //   gross = (102-100)*1 + (101-99)*1 = 2 + 2 = 4
        //   fees  = 0.10 + 0.11 + 0.12 + 0.09 = 0.42
        //   funding = 0.05 + (-0.02) = 0.03
        //   net   = gross - fees + funding = 4 - 0.42 + 0.03 = 3.61
        //   notional = 1 * (100+101)/2 = 100.5
        //   pct   = 3.61 / 100.5 * 100
        let gross: Decimal = t.gross_pnl.parse().unwrap();
        let fees: Decimal  = t.fees.parse().unwrap();
        let net: Decimal   = t.net_pnl.parse().unwrap();
        assert_eq!(gross, decm!(4));
        assert_eq!(fees, decm!(0.42));
        assert_eq!(net, decm!(3.61));
        // Accept a ~18-decimal-place diff on pct because printf('%.18f') on
        // the fees subquery introduces sub-cent FP noise on the SUM side;
        // the key invariant (net == gross - fees + funding) is already
        // asserted bit-exact above.
        let pct: Decimal = t.net_pnl_pct.parse().unwrap();
        let expected_pct = decm!(3.61) / decm!(100.5) * decm!(100);
        assert!((pct - expected_pct).abs() < decm!(0.0001),
                "pct {pct} vs expected {expected_pct}");
    }

    /// Verify the CTE path used by `load_all_time_stats`: aggregates over
    /// multiple trades and returns per-bucket totals with the same
    /// derivation as the per-trade loader.
    #[test]
    fn all_time_stats_derives_aggregates_from_orders() {
        let db = Storage::open(":memory:").expect("open in-memory");
        db.conn.execute(
            "INSERT INTO sessions (id, started_at_ms, state) VALUES (1, 1, 'active')",
            [],
        ).unwrap();

        let insert_trade = |id: &str, is_long_1: i64, funding_1: &str, funding_2: &str| {
            db.conn.execute(
                "INSERT INTO trades (
                    id, session_id, session_seq, symbol, is_long_1,
                    exchange_id_1, exchange_id_2, qty, state,
                    opened_at_ms, closed_at_ms, signal_at_ms, exit_reason,
                    funding_1, funding_2
                 ) VALUES (?1, 1, 1, 'BTC', ?2, 1, 2, '1', 'closed',
                           1, 2, 1, 'Tp', ?3, ?4)",
                params![id, is_long_1, funding_1, funding_2],
            ).unwrap();
        };
        let insert_order = |oid: u64, tid: &str, exchange_id: i64, purpose: &str,
                            avg_price: &str, fee: &str| {
            db.conn.execute(
                "INSERT INTO orders (
                    order_id, trade_id, session_seq, symbol, exchange_id, side,
                    order_type, purpose, qty, limit_price, status,
                    filled_qty, avg_fill_price, fee,
                    first_fill_at_ms, last_fill_at_ms, closed_at_ms, placed_at_ms
                ) VALUES (?1, ?2, 1, 'BTC', ?3, 'buy', 'ioc', ?4,
                          '1', ?5, 'filled',
                          '1', ?5, ?6, 1, 1, 1, 1)",
                params![oid, tid, exchange_id, purpose, avg_price, fee],
            ).unwrap();
        };

        // Trade A: winner. gross=+4, fees=0.4, funding=0 → net=+3.6
        insert_trade("a", 1, "0", "0");
        insert_order(1, "a", 1, "entry_1", "100", "0.10");
        insert_order(2, "a", 2, "entry_2", "101", "0.10");
        insert_order(3, "a", 1, "exit_1",  "102", "0.10");
        insert_order(4, "a", 2, "exit_2",  " 99", "0.10");

        // Trade B: loser. gross=-2, fees=0.2, funding=+0.1 → net=-2.1
        insert_trade("b", 1, "0.05", "0.05");
        insert_order(5, "b", 1, "entry_1", "100", "0.05");
        insert_order(6, "b", 2, "entry_2", "100", "0.05");
        insert_order(7, "b", 1, "exit_1",  " 99", "0.05");
        insert_order(8, "b", 2, "exit_2",  "101", "0.05");

        let stats = db.load_all_time_stats().expect("stats");
        assert_eq!(stats.trade_count, 2);
        assert_eq!(stats.wins, 1);
        assert_eq!(stats.losses, 1);
        // total_pnl is an f64 SUM from SQLite CTE; allow epsilon for FP.
        let expected_total = 3.6 + (-2.1);
        assert!((stats.total_pnl - expected_total).abs() < 1e-9,
                "total_pnl {} vs {}", stats.total_pnl, expected_total);
        let expected_fees = 0.4 + 0.2;
        assert!((stats.total_fees - expected_fees).abs() < 1e-9,
                "total_fees {} vs {}", stats.total_fees, expected_fees);
    }

    // ── Shared fixture helpers for the session-aggregate tests below ────────
    //
    // The existing `all_time_stats_derives_aggregates_from_orders` fixture
    // hardcodes session_id=1 / state='closed', so we define richer helpers
    // here that take the session_id and state, and let callers insert
    // sessions + partial-fill `failed_open` trades.

    fn insert_session(db: &Storage, id: i64, started_at_ms: i64) {
        db.conn.execute(
            "INSERT INTO sessions (id, started_at_ms, state) VALUES (?1, ?2, 'active')",
            params![id, started_at_ms],
        ).unwrap();
    }

    #[allow(clippy::too_many_arguments)]
    fn insert_trade_full(
        db: &Storage,
        id: &str, session_id: i64, state: &str, is_long_1: i64,
        qty: &str, funding_1: &str, funding_2: &str,
    ) {
        db.conn.execute(
            "INSERT INTO trades (
                id, session_id, session_seq, symbol, is_long_1,
                exchange_id_1, exchange_id_2, qty, state,
                opened_at_ms, closed_at_ms, signal_at_ms, exit_reason,
                funding_1, funding_2
             ) VALUES (?1, ?2, 1, 'BTC', ?3, 1, 2, ?4, ?5,
                       1, 2, 1, 'Tp', ?6, ?7)",
            params![id, session_id, is_long_1, qty, state, funding_1, funding_2],
        ).unwrap();
    }

    fn insert_order_row(db: &Storage, oid: u64, tid: &str, exchange_id: i64,
                        purpose: &str, avg_price: &str, fee: &str) {
        db.conn.execute(
            "INSERT INTO orders (
                order_id, trade_id, session_seq, symbol, exchange_id, side,
                order_type, purpose, qty, limit_price, status,
                filled_qty, avg_fill_price, fee,
                first_fill_at_ms, last_fill_at_ms, closed_at_ms, placed_at_ms
            ) VALUES (?1, ?2, 1, 'BTC', ?3, 'buy', 'ioc', ?4,
                      '1', ?5, 'filled',
                      '1', ?5, ?6, 1, 1, 1, 1)",
            params![oid, tid, exchange_id, purpose, avg_price, fee],
        ).unwrap();
    }

    /// Full 4-order "closed" trade: entry + exit on both legs.
    #[allow(clippy::too_many_arguments)]
    fn insert_closed_trade(
        db: &Storage, id: &str, session_id: i64, oid_base: u64,
        qty: &str, e1: &str, e2: &str, x1: &str, x2: &str, per_leg_fee: &str,
    ) {
        insert_trade_full(db, id, session_id, "closed", 1, qty, "0", "0");
        insert_order_row(db, oid_base,     id, 1, "entry_1", e1, per_leg_fee);
        insert_order_row(db, oid_base + 1, id, 2, "entry_2", e2, per_leg_fee);
        insert_order_row(db, oid_base + 2, id, 1, "exit_1",  x1, per_leg_fee);
        insert_order_row(db, oid_base + 3, id, 2, "exit_2",  x2, per_leg_fee);
    }

    /// Partial-fill `failed_open` trade: only entry_1 filled. This is the
    /// exact prod shape that produced the +$89 phantom PnL.
    fn insert_failed_open_trade(
        db: &Storage, id: &str, session_id: i64, oid: u64,
        qty: &str, e1: &str, fee: &str,
    ) {
        insert_trade_full(db, id, session_id, "failed_open", 0, qty, "0", "0");
        insert_order_row(db, oid, id, 1, "entry_1", e1, fee);
    }

    /// Regression: prod session #3 had a `failed_open` trade where only
    /// `entry_1` filled (EIGEN, qty=91, price≈0.978). The old SQL CTE
    /// computed `(0 - 0.97796) * 91 ≈ -88.99`, then the sign flipped by
    /// `is_long_1` and produced +$88.99 of phantom net PnL on a $15 stake.
    /// After routing aggregates through `derive_pnl`, the failed_open
    /// contributes net = -fees only. This test pins the regression.
    #[test]
    fn load_all_time_stats_excludes_failed_open_phantom_gross() {
        let db = Storage::open(":memory:").expect("open in-memory");
        insert_session(&db, 1, 1_000);

        // Four fully-filled closed trades (winners + loser).
        // All is_long_1=1, qty=1, per-leg fee=0.1 → total fees/trade = 0.4.
        insert_closed_trade(&db, "a", 1, 1, "1", "100", "101", "102", " 99", "0.10");
        //   gross = (102-100)*1 + (101-99)*1 = 4; net = 4 - 0.4 = 3.6
        insert_closed_trade(&db, "b", 1, 5, "1", "100", "100", "101", " 99", "0.10");
        //   gross = (101-100)*1 + (100-99)*1 = 2; net = 2 - 0.4 = 1.6
        insert_closed_trade(&db, "c", 1, 9, "1", "100", "100", " 99", "101", "0.10");
        //   gross = (99-100)*1 + (100-101)*1 = -2; net = -2 - 0.4 = -2.4
        insert_closed_trade(&db, "d", 1, 13, "1", "100", "100", "101", "100", "0.10");
        //   gross = (101-100)*1 + (100-100)*1 = 1; net = 1 - 0.4 = 0.6

        // The poisoned fixture: EIGEN-shaped failed_open, only entry_1
        // filled. qty=91, price=0.97796, fee=0.003300615.
        insert_failed_open_trade(&db, "eigen", 1, 17, "91", "0.97796", "0.003300615");

        let stats = db.load_all_time_stats().expect("stats");

        // Expected totals: derive_pnl over all five trades.
        let expected_total = 3.6_f64 + 1.6 - 2.4 + 0.6 - 0.003300615;
        let expected_fees  = 4.0 * 0.4 + 0.003300615;
        assert_eq!(stats.trade_count, 5);
        assert_eq!(stats.wins, 3, "a/b/d win");
        assert_eq!(stats.losses, 2, "c + failed_open lose");
        assert!((stats.total_pnl - expected_total).abs() < 1e-9,
                "total_pnl {} vs {}", stats.total_pnl, expected_total);
        assert!((stats.total_fees - expected_fees).abs() < 1e-9,
                "total_fees {} vs {}", stats.total_fees, expected_fees);

        // MUST NOT contain the phantom +88.99 from the failed_open leg.
        let phantom = 0.97796_f64 * 91.0;
        assert!((stats.total_pnl - phantom).abs() > 10.0,
                "total_pnl unexpectedly close to phantom {} (actual {})", phantom, stats.total_pnl);

        // Single session → best == worst == total.
        assert_eq!(stats.best_session_id, Some(1));
        assert_eq!(stats.worst_session_id, Some(1));
        assert!((stats.best_session_pnl - expected_total).abs() < 1e-9);
        assert!((stats.worst_session_pnl - expected_total).abs() < 1e-9);
    }

    /// Multisession cross-check: per-session totals must sum to all-time
    /// total. This is the structural no-drift invariant — if it ever
    /// fails, the aggregate path has diverged from `derive_pnl`.
    #[test]
    fn load_all_sessions_summary_handles_failed_open_and_multisession() {
        let db = Storage::open(":memory:").expect("open in-memory");
        insert_session(&db, 1, 1_000);
        insert_session(&db, 2, 2_000);

        // S1: one closed winner + one failed_open partial.
        insert_closed_trade(&db, "s1_win", 1, 1, "1", "100", "100", "102", " 98", "0.10");
        //   gross = 2 + 2 = 4; fees = 0.4; net = 3.6
        insert_failed_open_trade(&db, "s1_fail", 1, 5, "50", "10", "0.05");
        //   gross = 0; net = -0.05

        // S2: single loser.
        insert_closed_trade(&db, "s2_lose", 2, 9, "1", "100", "100", " 99", "101", "0.10");
        //   gross = -1 + -1 = -2; fees = 0.4; net = -2.4

        let summary = db.load_all_sessions_summary().expect("summary");
        assert_eq!(summary.len(), 2, "two sessions expected");
        // Ordered by started_at_ms DESC → S2 first.
        assert_eq!(summary[0].id, 2);
        assert_eq!(summary[1].id, 1);

        // S1 row: winner + failed_open.
        let s1 = &summary[1];
        assert_eq!(s1.trade_count, 2);
        assert_eq!(s1.wins, 1);
        assert_eq!(s1.losses, 1);
        let s1_expected = 3.6 + (-0.05);
        assert!((s1.total_pnl - s1_expected).abs() < 1e-9,
                "s1.total_pnl {} vs {}", s1.total_pnl, s1_expected);

        // S2 row: single loser.
        let s2 = &summary[0];
        assert_eq!(s2.trade_count, 1);
        assert_eq!(s2.wins, 0);
        assert_eq!(s2.losses, 1);
        assert!((s2.total_pnl - (-2.4)).abs() < 1e-9,
                "s2.total_pnl {} vs -2.4", s2.total_pnl);

        // No-drift invariant: sum of session totals == all-time total.
        let stats = db.load_all_time_stats().expect("stats");
        let sum = summary.iter().map(|r| r.total_pnl).sum::<f64>();
        assert!((sum - stats.total_pnl).abs() < 1e-9,
                "drift: sum(sessions)={} vs all_time={}", sum, stats.total_pnl);

        assert_eq!(stats.trade_count, 3);
        assert_eq!(stats.wins, 1);
        assert_eq!(stats.losses, 2);
    }

    /// Best/worst session resolution must come from folded `derive_pnl`
    /// totals, not from the dropped `trades.net_pnl` column.
    #[test]
    fn all_time_stats_best_worst_distinguish_sessions() {
        let db = Storage::open(":memory:").expect("open in-memory");
        insert_session(&db, 1, 1_000);
        insert_session(&db, 2, 2_000);
        insert_session(&db, 3, 3_000);

        // S1 ~ +3.6, S2 ~ -2.4, S3 ~ +0.6 → best=S1, worst=S2.
        insert_closed_trade(&db, "s1_a", 1, 1,  "1", "100", "100", "102", " 98", "0.10");
        insert_closed_trade(&db, "s2_a", 2, 5,  "1", "100", "100", " 99", "101", "0.10");
        insert_closed_trade(&db, "s3_a", 3, 9,  "1", "100", "100", "101", "100", "0.10");

        let stats = db.load_all_time_stats().expect("stats");
        assert_eq!(stats.best_session_id, Some(1));
        assert!((stats.best_session_pnl - 3.6).abs() < 1e-9,
                "best_session_pnl {}", stats.best_session_pnl);
        assert_eq!(stats.worst_session_id, Some(2));
        assert!((stats.worst_session_pnl - (-2.4)).abs() < 1e-9,
                "worst_session_pnl {}", stats.worst_session_pnl);
    }

    // ── external leg persistence (BUG B regression guards) ─────────────────

    /// Shape from the 2026-04-18 MEGA liquidation: entry + exit on
    /// exchange 1 via liquidation, entry + surviving-leg close on
    /// exchange 2. Without `persist_external_leg_fill`, only the
    /// surviving leg's exit row ever exists → derive_pnl sees exit_1=0
    /// → gross=0. With the helper, gross matches the hand-computed value.
    #[test]
    fn persist_external_leg_fill_lets_derive_pnl_compute_gross() {
        let db = Storage::open(":memory:").expect("open in-memory");
        insert_session(&db, 1, 1_000);

        // is_long_1 = false (Lighter short / HL long), qty=72.
        insert_trade_full(&db, "liq1", 1, "closed", 0, "72", "0", "0");
        // Entries from the real trade.
        insert_order_row(&db, 100, "liq1", 1, "entry_1", "0.2078",  "0.00419");
        insert_order_row(&db, 101, "liq1", 2, "entry_2", "0.20649", "0.00669");
        // Surviving HL leg was closed by the bot.
        insert_order_row(&db, 102, "liq1", 2, "exit_unwind_2", "0.22797", "0.00739");
        // BEFORE: without the external-leg row, derive_pnl returns gross=0.
        {
            let trades = db.load_closed_trades_for_session(1).expect("load");
            assert_eq!(trades.len(), 1);
            let gross: Decimal = trades[0].gross_pnl.parse().unwrap();
            assert_eq!(gross, Decimal::ZERO, "pre-fix UI state: gross==0");
        }
        // Record the external leg (Lighter liquidation @ 0.24960).
        db.persist_external_leg_fill(
            "liq1", 103, "MEGA", 1, "exit_unwind_1", "buy",
            Decimal::from_str("72").unwrap(),
            Decimal::from_str("0.24960").unwrap(),
            Decimal::from_str("0.00898").unwrap(),
            1_000_000,
            "liquidation",
        ).expect("persist external leg");

        let trades = db.load_closed_trades_for_session(1).expect("load");
        assert_eq!(trades.len(), 1);
        let t = &trades[0];
        // is_long_1=false: gross = (e1 - x1)*q + (x2 - e2)*q
        //   = (0.2078 - 0.24960)*72 + (0.22797 - 0.20649)*72
        //   = -3.0096 + 1.54656 = -1.46304
        let gross: Decimal = t.gross_pnl.parse().unwrap();
        let expected_gross = decm!(-1.46304);
        assert!(
            (gross - expected_gross).abs() < decm!(0.00001),
            "gross {gross} vs expected {expected_gross}"
        );
        // fees = 0.00419 + 0.00669 + 0.00739 + 0.00898 = 0.02725
        let fees: Decimal = t.fees.parse().unwrap();
        assert_eq!(fees, decm!(0.02725));
    }

    /// The helper must insert a row whose `purpose` matches the
    /// `purpose LIKE 'exit_%'` SELECT used by derive_pnl. Guards against
    /// silently picking a non-matching name (e.g. "external_close") which
    /// would compile but leave the UI broken.
    #[test]
    fn persist_external_leg_fill_row_is_picked_up_by_exit_select() {
        let db = Storage::open(":memory:").expect("open in-memory");
        insert_session(&db, 1, 1_000);
        insert_trade_full(&db, "t1", 1, "closed", 1, "1", "0", "0");
        insert_order_row(&db, 1, "t1", 1, "entry_1", "100", "0");
        insert_order_row(&db, 2, "t1", 2, "entry_2", "101", "0");
        insert_order_row(&db, 3, "t1", 1, "exit_1",  "102", "0");
        db.persist_external_leg_fill(
            "t1", 4, "BTC", 2, "exit_unwind_2", "buy",
            decm!(1), decm!(99.5), decm!(0), 1_000, "external_sl",
        ).unwrap();

        let trades = db.load_closed_trades_for_session(1).expect("load");
        assert_eq!(trades.len(), 1);
        // exit_2 must reflect the synthetic fill (0.5 below entry_2 → 1.5 gross on leg2).
        let gross: Decimal = trades[0].gross_pnl.parse().unwrap();
        // is_long_1=true: gross = (x1-e1)*q + (e2-x2)*q = (102-100)*1 + (101-99.5)*1 = 2 + 1.5 = 3.5
        assert_eq!(gross, decm!(3.5));
    }

    // ── session_start_balances (balance-delta sanity check) ───────────────

    #[test]
    fn session_start_balances_roundtrip_and_ordered_by_exchange_id() {
        let db = Storage::open(":memory:").expect("open in-memory");
        insert_session(&db, 42, 1_000);
        db.insert_session_start_balance(42, 2, decm!(40.11), 1_001).unwrap();
        db.insert_session_start_balance(42, 1, decm!(45.21), 1_001).unwrap();
        db.insert_session_start_balance(42, 4, decm!(46.79), 1_001).unwrap();
        db.insert_session_start_balance(42, 3, decm!(49.12), 1_001).unwrap();

        let balances = db.load_session_start_balances(42).expect("load");
        // Ordered by exchange_id.
        assert_eq!(balances.len(), 4);
        assert_eq!(balances[0].0, 1);
        assert_eq!(balances[0].1, decm!(45.21));
        assert_eq!(balances[1].0, 2);
        assert_eq!(balances[1].1, decm!(40.11));
        assert_eq!(balances[2].0, 3);
        assert_eq!(balances[2].1, decm!(49.12));
        assert_eq!(balances[3].0, 4);
        assert_eq!(balances[3].1, decm!(46.79));
    }

    #[test]
    fn session_start_balance_insert_or_ignore_preserves_first_snapshot() {
        let db = Storage::open(":memory:").expect("open in-memory");
        insert_session(&db, 1, 1_000);
        // First snapshot — the real session-start baseline.
        db.insert_session_start_balance(1, 1, decm!(10.00), 100).unwrap();
        // Bot restart within the same session — re-snapshot must NOT
        // overwrite, otherwise Balance Δ rebases every restart and hides
        // accumulated drift.
        db.insert_session_start_balance(1, 1, decm!(20.00), 200).unwrap();

        let balances = db.load_session_start_balances(1).unwrap();
        assert_eq!(balances.len(), 1);
        assert_eq!(balances[0].1, decm!(10.00));
    }

    #[test]
    fn session_start_balances_empty_for_unknown_session() {
        let db = Storage::open(":memory:").expect("open in-memory");
        // A session that predates the feature.
        insert_session(&db, 99, 1_000);
        let balances = db.load_session_start_balances(99).unwrap();
        assert!(balances.is_empty());
    }

    // ── Reconciliation DB recording (BUG A regression guards) ─────────────

    /// Reconciliation ORPHAN path: DB has a trade whose peer leg is gone
    /// from the other exchange. Bot closes the surviving leg via the
    /// trait; `record_reconciliation_exit` inserts the `exit_unwind_N`
    /// order row so `derive_pnl` sees a non-zero exit. Without this the
    /// trade shows gross=0 in the UI even though we actually closed it.
    #[test]
    fn record_reconciliation_exit_populates_exit_price_for_existing_trade() {
        let db = Storage::open(":memory:").expect("open in-memory");
        insert_session(&db, 1, 1_000);
        insert_trade_full(&db, "orph1", 1, "closed", 1, "5", "0", "0");
        insert_order_row(&db, 10, "orph1", 1, "entry_1", "100", "0.05");
        insert_order_row(&db, 11, "orph1", 2, "entry_2", "101", "0.05");
        // Pre-fix: only entry rows on file → derive_pnl gross=0.
        {
            let trades = db.load_closed_trades_for_session(1).expect("load");
            let gross: Decimal = trades[0].gross_pnl.parse().unwrap();
            assert_eq!(gross, Decimal::ZERO);
        }

        // Reconciliation closed the surviving leg_1 at $103 — bot's fill
        // knowledge from the close. Record it.
        db.record_reconciliation_exit(
            "orph1", 12, "BTC", 1, "exit_unwind_1", "sell",
            decm!(5), decm!(103), decm!(0.05), 2_000,
            "reconcile_orphan",
        ).unwrap();

        let trades = db.load_closed_trades_for_session(1).expect("load");
        let _gross: Decimal = trades[0].gross_pnl.parse().unwrap();
        // is_long_1=true, leg_2 exit missing (=0) → both_filled=false → 0.
        // Exit only on leg_1 isn't enough — must have both. This test
        // asserts the wire contract: record_reconciliation_exit inserts
        // the correct purpose, and derive_pnl picks it up on the right leg.
        // Insert the leg_2 exit too, to land a non-zero gross.
        db.record_reconciliation_exit(
            "orph1", 13, "BTC", 2, "exit_unwind_2", "buy",
            decm!(5), decm!(99), decm!(0.05), 2_000,
            "reconcile_orphan",
        ).unwrap();
        let trades = db.load_closed_trades_for_session(1).expect("load");
        let gross: Decimal = trades[0].gross_pnl.parse().unwrap();
        // (103-100)*5 + (101-99)*5 = 15 + 10 = 25
        assert_eq!(gross, decm!(25));
        let _ = gross;
    }

    /// Reconciliation LEAK path: no DB trade exists. `record_orphan_cover_trade`
    /// creates a synthetic trade with entry@entry_price → exit@cover_price,
    /// flagged with `exit_reason='orphan_cover'`. The UI aggregates it via
    /// `derive_pnl` so the loss shows up as part of Total PnL.
    #[test]
    fn record_orphan_cover_trade_appears_as_loss_in_all_time_stats() {
        let db = Storage::open(":memory:").expect("open in-memory");
        insert_session(&db, 1, 1_000);
        // MEGA: we had a naked −170 SHORT on exchange 4 (HL), entered at
        // avg ~0.1764. MEGA spiked, we covered at 0.17495 on restart.
        // was_long=false (short), orphan exchange=4, hedge synthetic=1.
        let tid = db.record_orphan_cover_trade(
            /*session_id*/ 1, "MEGA",
            /*orphan_exchange_id*/ 4, /*hedge_exchange_id*/ 1,
            /*qty*/ decm!(170), /*was_long*/ false,
            /*entry_price*/ decm!(0.1764),
            /*cover_price*/ decm!(0.17495),
            /*cover_fee*/ decm!(0.013),
            /*closed_at_ms*/ 1_000,
            100, 101, 102, 103,
        ).expect("record synthetic");
        assert!(!tid.is_empty());

        let stats = db.load_all_time_stats().expect("stats");
        assert_eq!(stats.trade_count, 1);
        // is_long_1 = was_long = false → gross = (e1 - x1)*q + (x2 - e2)*q
        //   = (0.1764 - 0.17495)*170 + (0.1764 - 0.1764)*170
        //   = 0.2465 + 0 = 0.2465
        // (short winning: covered below entry.)
        // net = gross - fee + funding = 0.2465 - 0.013 + 0 = 0.2335
        let expected_net = 0.2465 - 0.013;
        assert!(
            (stats.total_pnl - expected_net).abs() < 1e-6,
            "total_pnl {} vs expected {}", stats.total_pnl, expected_net
        );
    }

    /// Atomicity: `record_orphan_cover_trade` wraps trade + 4 orders in
    /// a single transaction. If a later INSERT fails (e.g. FK violation,
    /// duplicate order_id), the earlier trade row must be rolled back —
    /// otherwise `derive_pnl` would see a trades row with missing entries
    /// and silently contribute gross=0 to Total PnL.
    #[test]
    fn record_orphan_cover_trade_rolls_back_on_failure() {
        let db = Storage::open(":memory:").expect("open in-memory");
        insert_session(&db, 1, 1_000);
        // Force a collision: insert an order with id=201 first. Then the
        // orphan_cover attempt, which tries to use the same `orphan_order_id`,
        // will collide on INSERT OR REPLACE (replacing an order belonging
        // to a non-existent trade). We rely on the FK on `orders.trade_id`
        // failing after the synthetic trade is inserted — actually simpler:
        // use a non-existent session_id so the trades FK fails.
        let err = db.record_orphan_cover_trade(
            /*session_id*/ 999_999, // no such session → FK violation on trades
            "TEST", 4, 1, decm!(10), true, decm!(100), decm!(95), decm!(0.5),
            1_000, 500, 501, 502, 503,
        );
        assert!(err.is_err(), "expected FK violation to bubble up");
        // After rollback the trades table must still be empty.
        let count: i64 = db.conn
            .query_row("SELECT COUNT(*) FROM trades WHERE exit_reason = 'orphan_cover'", [], |r| r.get(0))
            .unwrap();
        assert_eq!(count, 0, "trade row must be rolled back after inner INSERT failure");
        let order_count: i64 = db.conn
            .query_row("SELECT COUNT(*) FROM orders WHERE order_id IN (500, 501, 502, 503)", [], |r| r.get(0))
            .unwrap();
        assert_eq!(order_count, 0, "no orphaned order rows after rollback");
    }

    /// Single-exchange config fallback: when `pick_hedge_exchange_id`
    /// returns the same id as orphan (only one exchange configured), the
    /// `AGG_TRADES_SELECT` result is inaccurate in TWO ways:
    ///   1. The LEFT JOIN `oe1`/`oe2` match both entry rows on the
    ///      single exchange → cartesian product produces 4 rows per
    ///      trade instead of 1.
    ///   2. The `exit_1` / `exit_2` subqueries both filter by the same
    ///      `exchange_id` (=orphan) and return the SAME `ORDER BY ...
    ///      LIMIT 1` row — so entry_1, entry_2, exit_1, exit_2 all
    ///      collapse to the same prices. gross ends up 0 per row, and
    ///      net = −fees × 4 (fees are deduplicated by the subquery).
    ///
    /// Net result for the test below: gross=0, fees=0.5, net_per_row=-0.5,
    /// total across 4 cartesian rows = -2.0.
    ///
    /// This test locks the CURRENT behavior so a future fix that
    /// collapses the duplication (via DISTINCT or GROUP BY) is detected
    /// as a behavior change — not a silent regression. Production runs
    /// with 4 exchanges so this edge case is warned about in
    /// `pick_hedge_exchange_id` and not observed in practice.
    #[test]
    fn record_orphan_cover_trade_single_exchange_is_best_effort() {
        let db = Storage::open(":memory:").expect("open in-memory");
        insert_session(&db, 1, 1_000);
        // orphan and hedge on the SAME exchange id (single-exchange fallback).
        let _tid = db.record_orphan_cover_trade(
            1, "XX", 1, 1, decm!(10), true, decm!(100), decm!(95), decm!(0.5),
            1_000, 600, 601, 602, 603,
        ).expect("record");

        let stats = db.load_all_time_stats().expect("stats");
        // Current behavior: gross=0, net = -fees × 4 = -2. This is not
        // the "true" loss (which would be gross = (95-100)*10 = -50,
        // net = -50.5) — the single-exchange layout makes the exit
        // subqueries return the orphan-cover row for BOTH legs, so
        // gross cancels to zero. Documented trade-off, acceptable only
        // for the degenerate single-exchange setup.
        assert!(
            (stats.total_pnl - (-2.0)).abs() < 1e-6,
            "single-exchange orphan_cover best-effort net is -2 (fees×4); got total_pnl={}",
            stats.total_pnl
        );
    }

    /// Symmetric: orphan was LONG and cover price fell → loss.
    /// Models the 2026-04-18 scenario where the HL orphan actually
    /// worked out neutral or profitable for HL short case, but this test
    /// covers the LONG direction with a real loss.
    #[test]
    fn record_orphan_cover_trade_long_orphan_loss() {
        let db = Storage::open(":memory:").expect("open in-memory");
        insert_session(&db, 1, 1_000);
        let tid = db.record_orphan_cover_trade(
            1, "BTC", 4, 1,
            decm!(10), /*was_long*/ true,
            /*entry*/ decm!(100), /*cover*/ decm!(95),
            /*fee*/ decm!(0.5),
            1_000, 200, 201, 202, 203,
        ).expect("record");
        assert!(!tid.is_empty());

        let stats = db.load_all_time_stats().expect("stats");
        // is_long_1=true: gross = (95-100)*10 + (100-100)*10 = -50
        // net = -50 - 0.5 = -50.5
        let expected = -50.0 - 0.5;
        assert!(
            (stats.total_pnl - expected).abs() < 1e-6,
            "total_pnl {} vs expected {}", stats.total_pnl, expected
        );
    }

    // ── load_last_ended_session (clean-mode orphan attribution) ──────────

    #[test]
    fn load_last_ended_session_returns_none_when_no_ended() {
        let db = Storage::open(":memory:").expect("open in-memory");
        // Only an active session present.
        insert_session(&db, 1, 1_000);
        let got = db.load_last_ended_session().expect("query");
        assert_eq!(got, None);
    }

    #[test]
    fn load_last_ended_session_returns_latest_ended_by_time() {
        let db = Storage::open(":memory:").expect("open in-memory");
        // Session 1: ended earliest
        insert_session(&db, 1, 1000);
        db.conn.execute(
            "UPDATE sessions SET state='ended', ended_at_ms=2000, end_reason='first' WHERE id=1",
            [],
        ).unwrap();
        // Session 2: ended most recent
        insert_session(&db, 2, 3000);
        db.conn.execute(
            "UPDATE sessions SET state='ended', ended_at_ms=5000, end_reason='latest' WHERE id=2",
            [],
        ).unwrap();
        // Session 3: still active — must be ignored
        insert_session(&db, 3, 6000);

        let got = db.load_last_ended_session().expect("query");
        assert_eq!(got, Some(2), "most recent ended (by ended_at_ms) should win");
    }

    /// End-to-end clean-mode scenario: previous session has an active
    /// trade. On clean restart, a new session starts, and the residual
    /// position from the old session gets closed via
    /// `record_reconciliation_exit` writing an `exit_unwind_1` order
    /// against the old trade. Assert:
    ///   1. The new exit row carries the OLD session_id (via trade FK).
    ///   2. `load_all_time_stats` counts the loss against session 1.
    ///   3. `session_start_balances` for the new session stays pristine
    ///      (no rows inserted here — they're the main.rs job).
    #[test]
    fn clean_mode_fill_lands_on_prior_session() {
        let db = Storage::open(":memory:").expect("open in-memory");
        insert_session(&db, 1, 1_000);
        db.conn.execute(
            "UPDATE sessions SET state='ended', ended_at_ms=2000 WHERE id=1",
            [],
        ).unwrap();
        insert_session(&db, 2, 3_000);

        // Active trade from session 1 (long, ex_1=1 ex_2=2 qty=5 entry=100/101).
        insert_trade_full(&db, "tA", 1, "active", 1, "5", "0", "0");
        insert_order_row(&db, 10, "tA", 1, "entry_1", "100", "0.05");
        insert_order_row(&db, 11, "tA", 2, "entry_2", "101", "0.05");

        // Simulate main.rs post-process: cleaned_fills contains fills for
        // the residual positions on both exchanges of session-1 trade.
        // Leg 1 closed at 105, leg 2 at 97.
        db.record_reconciliation_exit(
            "tA", 12, "BTC", 1, "exit_unwind_1", "sell",
            decm!(5), decm!(105), decm!(0.05), 4_000,
            "clean_mode_residual",
        ).unwrap();
        db.record_reconciliation_exit(
            "tA", 13, "BTC", 2, "exit_unwind_2", "buy",
            decm!(5), decm!(97), decm!(0.05), 4_000,
            "clean_mode_residual",
        ).unwrap();
        // Trade state flipped by main.rs close_all_active_trades.
        db.conn.execute(
            "UPDATE trades SET state='closed', closed_at_ms=4000 WHERE id='tA'",
            [],
        ).unwrap();

        // (1) Exit orders inherit trade.session_id = 1 (via FK) — verify SQL.
        let orders_session: i64 = db.conn.query_row(
            "SELECT t.session_id FROM orders o JOIN trades t ON o.trade_id = t.id
             WHERE o.order_id = 12",
            [],
            |r| r.get(0),
        ).unwrap();
        assert_eq!(orders_session, 1, "exit_unwind_1 inherits session 1");

        // (2) Per-session aggregates: session 1 should show the loss, not session 2.
        let summary = db.load_all_sessions_summary().expect("summary");
        let s1 = summary.iter().find(|s| s.id == 1).expect("session 1 row");
        let s2 = summary.iter().find(|s| s.id == 2);
        // is_long_1=true: gross = (105-100)*5 + (101-97)*5 = 25 + 20 = 45
        // fees = 0.05*4 = 0.2 → net = 44.8. Here the close shows a WIN.
        assert!(
            (s1.total_pnl - 44.8).abs() < 1e-6,
            "session 1 total_pnl should include the clean-mode exit: got {}",
            s1.total_pnl
        );
        // Session 2 was just created — no trades, zero pnl (or absent).
        if let Some(s2) = s2 {
            assert!(s2.total_pnl.abs() < 1e-9, "session 2 untouched: got {}", s2.total_pnl);
        }
    }

    /// When there is no matching DB trade for a cleaned position
    /// (true orphan — opened manually on exchange, or pre-tracking),
    /// the synthetic orphan_cover trade must be attributed to
    /// `load_last_ended_session()`, not the new session.
    #[test]
    fn orphan_clean_close_attributes_to_last_ended_session() {
        let db = Storage::open(":memory:").expect("open in-memory");
        insert_session(&db, 1, 1_000);
        db.conn.execute(
            "UPDATE sessions SET state='ended', ended_at_ms=2000 WHERE id=1",
            [],
        ).unwrap();
        insert_session(&db, 2, 3_000);

        let last = db.load_last_ended_session().unwrap().expect("have ended");
        assert_eq!(last, 1);

        // Synthesise an orphan_cover trade with the old session id.
        let _tid = db.record_orphan_cover_trade(
            last, "BTC", 4, 1,
            decm!(10), /*was_long*/ true,
            /*entry*/ decm!(100), /*cover*/ decm!(95), decm!(0.5),
            4_000, 300, 301, 302, 303,
        ).expect("record");

        // Verify trade is attributed to session 1, not 2.
        let sid: i64 = db.conn.query_row(
            "SELECT session_id FROM trades WHERE exit_reason = 'orphan_cover'",
            [],
            |r| r.get(0),
        ).unwrap();
        assert_eq!(sid, 1, "orphan_cover landed on prior session 1");
    }
}