hypercall/liquidator/

watcher.rs

//! Liquidation watcher background task.
//!
//! Polls accounts for liquidation health and drives both backend-managed
//! partial liquidation and on-chain full liquidation transitions.

use super::cache::LiquidationCache;
use super::executor::LiquidationExecutor;
use super::health_check::{check_portfolio_health, check_standard_health, LiquidationHealthResult};
use super::partial::{
    compute_partial_bonus_bps, greedy_partial_plan, is_above_partial_target, required_mm_relief,
    target_equity_from_mm, LiquidationSliceCandidate, PartialLiquidationOrderPlan,
};
use super::state::{
    has_material_projection_change, AccountLiquidationStatus, LiquidationState,
    PartialLiquidationMetadata,
};
use crate::portfolio::{PortfolioService, PositionData};
use crate::read_cache::tier::TierCache;
use crate::rsm::liquidation_manager::LiquidationManager;
use crate::rsm::margin_service::SpanMarginService;
use crate::rsm::unified_engine::{increment_pending_requests, UnifiedEngineRequest};
use crate::rsm::MarginMode;
use crate::shared::order_types::{to_contract_units_decimal, ParsedSymbol};
use crate::standard_margin::StandardAccountBuilder;
use anyhow::{anyhow, Context};
use hypercall_margin::standard::StandardAccount;
use hypercall_margin::{PortfolioMarginOptionKey, PortfolioMarginSnapshot};
use hypercall_runtime_api::{OrderSnapshotProvider, QuoteProvider, RuntimeOrderSummary};
use hypercall_types::WalletAddress;
use hypercall_types::{
    EngineMessage, LiquidationStateMessage, LiquidationStateType, OrderAction, OrderActionMessage,
    OrderInfo, OrderUpdateMessage, OrderUpdateStatus, Side, TimeInForce,
};
use metrics::{counter, gauge};
use rust_decimal::Decimal;
use rust_decimal_macros::dec;
use std::collections::{HashMap, HashSet};
use std::sync::Arc;
use std::time::{Duration, Instant};
use tokio::sync::broadcast;
use tokio::sync::mpsc;
use tokio::time::{interval, timeout};
use tracing::{debug, error, info, warn};
use uuid::Uuid;

const ENGINE_RESPONSE_TIMEOUT: Duration = Duration::from_secs(10);
const PARTIAL_LIQ_MIN_SIZE: Decimal = dec!(0.00000001);
const PARTIAL_PLAN_MAX_ROUNDS: usize = 3;

#[derive(Debug)]
struct PartialCycleOutcome {
    metadata: PartialLiquidationMetadata,
    can_recover: bool,
}

#[derive(Debug)]
struct OpenOrderSyncResult {
    open_liquidation_client_ids: Vec<String>,
}

#[derive(Debug, Clone)]
struct CancelTarget {
    order_id: u64,
    client_id: Option<String>,
}

/// Configuration for the liquidation watcher.
#[derive(Clone, Debug)]
pub struct LiquidationWatcherConfig {
    /// How often to poll accounts for health (default: 5 seconds).
    pub poll_interval: Duration,
    /// Full-liquidation escalation timeout after pre-liquidation begins.
    pub full_escalation_timeout: Duration,
    /// Minimum shortfall to trigger pre-liquidation (default: 0, any shortfall triggers).
    pub min_shortfall_threshold: Decimal,
    /// Health target buffer for partial liquidation sizing metadata.
    pub partial_target_buffer_bps: u32,
    /// Minimum time between partial-liquidation reprices.
    pub partial_reprice_interval: Duration,
    /// Starting liquidation bonus in basis points.
    pub partial_bonus_start_bps: u32,
    /// Maximum liquidation bonus in basis points.
    pub partial_bonus_max_bps: u32,
    /// Time for bonus ramp from start to max.
    pub partial_bonus_ramp_ms: u64,
}

impl Default for LiquidationWatcherConfig {
    fn default() -> Self {
        Self {
            poll_interval: Duration::from_secs(5),
            full_escalation_timeout: Duration::from_secs(60),
            min_shortfall_threshold: dec!(0),
            partial_target_buffer_bps: 500,
            partial_reprice_interval: Duration::from_secs(5),
            partial_bonus_start_bps: 50,
            partial_bonus_max_bps: 1_000,
            partial_bonus_ramp_ms: 300_000,
        }
    }
}

impl LiquidationWatcherConfig {
    pub fn from_runtime_config(config: &crate::backend_config::LiquidationRuntimeConfig) -> Self {
        Self {
            poll_interval: Duration::from_millis(config.health_poll_interval_ms),
            full_escalation_timeout: Duration::from_millis(config.full_escalation_timeout_ms),
            min_shortfall_threshold: config.min_shortfall_threshold,
            partial_target_buffer_bps: config.partial_target_buffer_bps,
            partial_reprice_interval: Duration::from_millis(config.partial_reprice_interval_ms),
            partial_bonus_start_bps: config.partial_bonus_start_bps,
            partial_bonus_max_bps: config.partial_bonus_max_bps,
            partial_bonus_ramp_ms: config.partial_bonus_ramp_ms,
        }
    }
}

/// Liquidation watcher background task.
///
/// Monitors accounts for margin health and triggers liquidation state transitions.
pub struct LiquidationWatcher {
    /// Liquidation state cache.
    cache: Arc<LiquidationCache>,
    /// Executed portfolio state used for liquidation health and sizing.
    portfolio_service: Arc<dyn PortfolioService + Send + Sync>,
    /// Tier cache for margin mode lookup.
    tier_cache: Option<Arc<TierCache>>,
    /// SPAN margin service for portfolio margin accounts.
    span_margin_service: Arc<SpanMarginService>,
    /// Standard margin service.
    standard_margin_service: Arc<crate::standard_margin::StandardMarginService>,
    /// Risk account builder for portfolio margin.
    risk_account_builder: Option<Arc<crate::rsm::portfolio_margin::RiskAccountBuilder>>,
    /// Standard account builder for standard margin.
    standard_account_builder: Option<Arc<StandardAccountBuilder>>,
    /// Quote provider for liquidation pricing (BBO from orderbook).
    quote_provider: Option<Arc<dyn QuoteProvider>>,
    /// Greeks cache for theoretical price fallback when BBO unavailable.
    greeks_cache: Option<Arc<crate::read_cache::greeks::GreeksCache>>,
    /// Open-order snapshot provider for cancel/reprice decisions.
    order_snapshot: Option<Arc<dyn OrderSnapshotProvider>>,
    /// Order sender for backend-generated partial liquidation orders.
    order_sender: Option<mpsc::Sender<UnifiedEngineRequest>>,
    /// Event sender for publishing state changes.
    event_sender: Option<mpsc::UnboundedSender<EngineMessage>>,
    /// Liquidation executor for onchain calls.
    executor: Option<Arc<LiquidationExecutor>>,
    /// Configuration.
    config: LiquidationWatcherConfig,
}

impl LiquidationWatcher {
    /// Create a new liquidation watcher.
    pub fn new(
        cache: Arc<LiquidationCache>,
        portfolio_service: Arc<dyn PortfolioService + Send + Sync>,
        tier_cache: Arc<TierCache>,
        span_margin_service: Arc<SpanMarginService>,
        standard_margin_service: Arc<crate::standard_margin::StandardMarginService>,
        config: LiquidationWatcherConfig,
    ) -> Self {
        Self {
            cache,
            portfolio_service,
            tier_cache: Some(tier_cache),
            span_margin_service,
            standard_margin_service,
            risk_account_builder: None,
            standard_account_builder: None,
            quote_provider: None,
            greeks_cache: None,
            order_snapshot: None,
            order_sender: None,
            event_sender: None,
            executor: None,
            config,
        }
    }

    #[cfg(test)]
    fn new_for_tests_without_tier_cache(
        cache: Arc<LiquidationCache>,
        portfolio_service: Arc<dyn PortfolioService + Send + Sync>,
        span_margin_service: Arc<SpanMarginService>,
        standard_margin_service: Arc<crate::standard_margin::StandardMarginService>,
        config: LiquidationWatcherConfig,
    ) -> Self {
        Self {
            cache,
            portfolio_service,
            tier_cache: None,
            span_margin_service,
            standard_margin_service,
            risk_account_builder: None,
            standard_account_builder: None,
            quote_provider: None,
            greeks_cache: None,
            order_snapshot: None,
            order_sender: None,
            event_sender: None,
            executor: None,
            config,
        }
    }

    /// Set the risk account builder for portfolio margin accounts.
    pub fn with_risk_account_builder(
        mut self,
        builder: Arc<crate::rsm::portfolio_margin::RiskAccountBuilder>,
    ) -> Self {
        self.risk_account_builder = Some(builder);
        self
    }

    /// Set the standard account builder.
    pub fn with_standard_account_builder(mut self, builder: Arc<StandardAccountBuilder>) -> Self {
        self.standard_account_builder = Some(builder);
        self
    }

    /// Set the quote provider used for liquidation repricing.
    pub fn with_quote_provider(mut self, provider: Arc<dyn QuoteProvider>) -> Self {
        self.quote_provider = Some(provider);
        self
    }

    /// Set the greeks cache for theoretical price fallback.
    pub fn with_greeks_cache(mut self, cache: Arc<crate::read_cache::greeks::GreeksCache>) -> Self {
        self.greeks_cache = Some(cache);
        self
    }

    /// Set the order snapshot provider used for liquidation cancels.
    pub fn with_order_snapshot(mut self, snapshot: Arc<dyn OrderSnapshotProvider>) -> Self {
        self.order_snapshot = Some(snapshot);
        self
    }

    /// Set the order sender used for backend-generated liquidation orders.
    pub fn with_order_sender(mut self, sender: mpsc::Sender<UnifiedEngineRequest>) -> Self {
        self.order_sender = Some(sender);
        self
    }

    /// Set the event sender for publishing state changes.
    pub fn with_event_sender(mut self, sender: mpsc::UnboundedSender<EngineMessage>) -> Self {
        self.event_sender = Some(sender);
        self
    }

    /// Set the liquidation executor for onchain calls.
    pub fn with_executor(mut self, executor: Arc<LiquidationExecutor>) -> Self {
        self.executor = Some(executor);
        self
    }

    /// Run the watcher loop.
    pub async fn run(&self) {
        info!(
            "Starting liquidation watcher with poll interval {:?}",
            self.config.poll_interval
        );

        let mut interval = interval(self.config.poll_interval);

        loop {
            interval.tick().await;

            if let Err(e) = self.check_all_accounts().await {
                error!("Error checking accounts for liquidation: {}", e);
            }
        }
    }

    pub async fn run_with_shutdown(&self, mut shutdown_rx: broadcast::Receiver<()>) {
        info!(
            "Starting liquidation watcher with poll interval {:?}",
            self.config.poll_interval
        );

        let mut interval = interval(self.config.poll_interval);

        loop {
            tokio::select! {
                _ = shutdown_rx.recv() => {
                    info!("Liquidation watcher received shutdown signal");
                    break;
                }
                _ = interval.tick() => {
                    if let Err(e) = self.check_all_accounts().await {
                        error!("Error checking accounts for liquidation: {}", e);
                    }
                }
            }
        }
    }

    /// Check all accounts for liquidation health.
    async fn check_all_accounts(&self) -> anyhow::Result<()> {
        let now = get_timestamp_millis();

        // Health polling is driven by executed portfolios, but wallets already in the
        // liquidation state machine must continue to be processed even after their last
        // short option is closed. Otherwise pre-liquidation can never clear.
        let all_portfolios = self.portfolio_service.all_portfolios().await;
        let mut wallets_to_check: HashSet<WalletAddress> = all_portfolios.keys().copied().collect();
        for wallet in self.cache.get_all_wallets().await {
            if matches!(
                self.cache.get_state(&wallet).await,
                LiquidationState::PreLiquidation(..) | LiquidationState::InLiquidation(..)
            ) {
                wallets_to_check.insert(wallet);
            }
        }

        let wallet_count = wallets_to_check.len();
        if wallet_count == 0 {
            gauge!("ht_liquidation_partial_active_orders").set(0.0);
            return Ok(());
        }

        debug!("Checking {} accounts for liquidation health", wallet_count);

        let mut pre_liq_count = 0;
        let mut healthy_count = 0;
        let mut active_partial_orders = 0usize;

        for wallet in wallets_to_check {
            let has_positions = all_portfolios
                .get(&wallet)
                .is_some_and(|portfolio| !portfolio.positions.is_empty());
            let Some(existing_status) = self.cache.get_status(&wallet).await else {
                if !has_positions {
                    continue;
                }
                // Status will be initialized after health is computed below.
                // Fall through for accounts discovered via the portfolio cache.
                let Ok(margin_mode) = self.margin_mode_for_wallet(&wallet).await else {
                    warn!(
                        "Failed to determine margin mode for newly discovered wallet {}",
                        wallet
                    );
                    continue;
                };
                let health_result = match margin_mode {
                    MarginMode::Portfolio => self.check_portfolio_margin_health(&wallet).await,
                    MarginMode::Standard => self.check_standard_margin_health(&wallet).await,
                };

                let health = match health_result {
                    Ok(h) => h,
                    Err(e) => {
                        warn!("Failed to check health for {}: {}", wallet, e);
                        continue;
                    }
                };

                self.process_health_check(&wallet, &health, margin_mode, now)
                    .await;

                if let Some(status) = self.cache.get_status(&wallet).await {
                    if let LiquidationState::PreLiquidation(metadata) = &status.state {
                        active_partial_orders += metadata.active_order_client_ids.len();
                    }
                }

                if health.is_liquidatable {
                    pre_liq_count += 1;
                } else {
                    healthy_count += 1;
                }
                continue;
            };
            if !has_positions && existing_status.state.is_healthy() {
                continue;
            }

            let Ok(margin_mode) = self.margin_mode_for_wallet(&wallet).await else {
                warn!(
                    "Failed to determine margin mode for tracked wallet {}",
                    wallet
                );
                continue;
            };
            let health_result = match margin_mode {
                MarginMode::Portfolio => self.check_portfolio_margin_health(&wallet).await,
                MarginMode::Standard => self.check_standard_margin_health(&wallet).await,
            };

            let health = match health_result {
                Ok(h) => h,
                Err(e) => {
                    warn!("Failed to check health for {}: {}", wallet, e);
                    continue;
                }
            };

            self.process_health_check(&wallet, &health, margin_mode, now)
                .await;

            if let Some(status) = self.cache.get_status(&wallet).await {
                if let LiquidationState::PreLiquidation(metadata) = &status.state {
                    active_partial_orders += metadata.active_order_client_ids.len();
                }
            }

            if health.is_liquidatable {
                pre_liq_count += 1;
            } else {
                healthy_count += 1;
            }
        }

        gauge!("ht_liquidation_partial_active_orders").set(active_partial_orders as f64);

        if pre_liq_count > 0 {
            info!(
                "Liquidation check complete: {} healthy, {} in/near pre-liquidation",
                healthy_count, pre_liq_count
            );
        }

        Ok(())
    }

    async fn margin_mode_for_wallet(&self, wallet: &WalletAddress) -> anyhow::Result<MarginMode> {
        if let Some(tier_cache) = &self.tier_cache {
            return tier_cache.get_margin_mode(wallet).await;
        }

        self.cache
            .get_status(wallet)
            .await
            .map(|status| status.margin_mode)
            .ok_or_else(|| {
                anyhow!(
                    "tier_cache not configured and wallet {} not tracked",
                    wallet
                )
            })
    }

    /// Check health for a portfolio margin account.
    async fn check_portfolio_margin_health(
        &self,
        wallet: &WalletAddress,
    ) -> anyhow::Result<LiquidationHealthResult> {
        let builder = self
            .risk_account_builder
            .as_ref()
            .ok_or_else(|| anyhow!("RiskAccountBuilder not set"))?;

        let account = builder
            .build_executed_account_for_risk(wallet)
            .await
            .map_err(|e| anyhow!("Failed to build risk account: {:?}", e))?;

        Ok(check_portfolio_health(
            &wallet.to_string(),
            &account,
            self.span_margin_service.as_ref(),
        )
        .await)
    }

    /// Check health for a standard margin account.
    async fn check_standard_margin_health(
        &self,
        wallet: &WalletAddress,
    ) -> anyhow::Result<LiquidationHealthResult> {
        let builder = self
            .standard_account_builder
            .as_ref()
            .ok_or_else(|| anyhow!("StandardAccountBuilder not set"))?;

        let account = builder
            .build(wallet)
            .await
            .map_err(|e| anyhow!("Failed to build standard account: {}", e))?;

        Ok(check_standard_health(
            &wallet.to_string(),
            &account,
            self.standard_margin_service.as_ref(),
        ))
    }

    /// Process health check result and trigger state transitions.
    async fn process_health_check(
        &self,
        wallet: &WalletAddress,
        health: &LiquidationHealthResult,
        margin_mode: MarginMode,
        now: u64,
    ) {
        let equity = decimal_from_f64(health.equity, "equity");
        let mm_required = decimal_from_f64(health.mm_required, "mm_required");
        let shortfall = decimal_from_f64(health.shortfall(), "shortfall");

        self.cache
            .init_if_absent(*wallet, margin_mode, equity, mm_required, now)
            .await;
        self.cache
            .update_health(wallet, equity, mm_required, now)
            .await;

        let Some(status) = self.cache.get_status(wallet).await else {
            return;
        };

        match status.state.clone() {
            LiquidationState::Healthy => {
                if health.is_liquidatable && shortfall >= self.config.min_shortfall_threshold {
                    let metadata = PartialLiquidationMetadata {
                        entered_at: now,
                        mm_shortfall: shortfall,
                        target_equity: target_equity_from_mm(
                            mm_required,
                            self.config.partial_target_buffer_bps,
                        ),
                        escalation_deadline: now
                            + self.config.full_escalation_timeout.as_millis() as u64,
                        last_reprice_at: None,
                        active_order_request_ids: Vec::new(),
                        active_order_client_ids: Vec::new(),
                        bonus_bps: self.config.partial_bonus_start_bps,
                        pending_full_auction_id: None,
                        pending_full_request_id: None,
                        pending_full_tx_hash: None,
                        pending_full_margin_needed: None,
                    };

                    if let Some(previous) = self
                        .cache
                        .enter_pre_liquidation(wallet, metadata, now)
                        .await
                    {
                        counter!("ht_liquidation_partial_enter_total").increment(1);
                        if let Some(updated_status) = self.cache.get_status(wallet).await {
                            self.emit_state_change(&previous, &updated_status, now);
                        }
                    }

                    self.handle_pre_liquidation_state(wallet, margin_mode, now)
                        .await;
                }
            }
            LiquidationState::PreLiquidation(..) => {
                self.handle_pre_liquidation_state(wallet, margin_mode, now)
                    .await;
            }
            LiquidationState::InLiquidation(metadata) => {
                if !health.is_liquidatable {
                    if metadata.stop_request_id.is_some() {
                        debug!(
                            "Wallet {} recovered while liquidation stop request is already pending",
                            wallet
                        );
                    } else if let Some(chain_start_time) = metadata.chain_start_time {
                        if let Some(executor) = &self.executor {
                            match executor.stop_auction(wallet, chain_start_time).await {
                                Ok(request_id) => {
                                    info!(
                                        "Enqueued stop liquidation request {} for recovered wallet {}",
                                        request_id, wallet
                                    );
                                }
                                Err(e) => {
                                    error!(
                                        "Failed to enqueue stop liquidation for wallet {}: {}",
                                        wallet, e
                                    );
                                }
                            }
                        }
                    } else {
                        debug!(
                            "Wallet {} recovered but liquidation has no observed chain_start_time yet",
                            wallet
                        );
                    }
                }
            }
            LiquidationState::Liquidated(..) => {}
        }
    }

    async fn handle_pre_liquidation_state(
        &self,
        wallet: &WalletAddress,
        margin_mode: MarginMode,
        now: u64,
    ) {
        let Some(status) = self.cache.get_status(wallet).await else {
            return;
        };

        let LiquidationState::PreLiquidation(mut metadata) = status.state.clone() else {
            return;
        };

        let original_metadata = metadata.clone();
        metadata.mm_shortfall = status.shortfall();
        metadata.target_equity =
            target_equity_from_mm(status.mm_required, self.config.partial_target_buffer_bps);
        let above_partial_target = is_above_partial_target(
            status.equity,
            status.mm_required,
            self.config.partial_target_buffer_bps,
        );

        if metadata.pending_full_request_id.is_some() {
            if now
                > metadata.escalation_deadline
                    + self.config.full_escalation_timeout.as_millis() as u64
            {
                warn!(
                    "Clearing stale pending_full_request_id for wallet {} (deadline+timeout exceeded)",
                    wallet
                );
                metadata.pending_full_auction_id = None;
                metadata.pending_full_request_id = None;
                metadata.pending_full_tx_hash = None;
                metadata.pending_full_margin_needed = None;
                // Persist cleared state to cache before escalation retry
                self.persist_pre_liquidation_metadata(wallet, metadata.clone(), now)
                    .await;
            } else {
                if metadata != original_metadata {
                    self.persist_pre_liquidation_metadata(wallet, metadata, now)
                        .await;
                }
                return;
            }
        }

        let has_short_option_positions = self.has_short_option_positions(wallet).await;
        if !above_partial_target
            && (status.equity < Decimal::ZERO
                || matches!(has_short_option_positions, Some(false))
                || now >= metadata.escalation_deadline)
        {
            if let Err(error) = self
                .sync_and_cancel_orders(wallet, &metadata.active_order_client_ids, true)
                .await
            {
                warn!(
                    "Failed to synchronize orders before full-liquidation escalation for {}: {}",
                    wallet, error
                );
            }

            if let Some(executor) = &self.executor {
                match executor.start_auction(wallet).await {
                    Ok(auction_id) => {
                        info!(
                            "Enqueued liquidation auction {} for wallet {} from pre-liquidation",
                            auction_id, wallet
                        );
                    }
                    Err(error) => {
                        error!(
                            "Failed to enqueue liquidation auction for wallet {}: {}",
                            wallet, error
                        );
                    }
                }
            } else {
                debug!(
                    "Wallet {} reached full-liquidation escalation but no executor is configured",
                    wallet
                );
            }
            return;
        }

        match self
            .run_partial_liquidation_cycle(wallet, margin_mode, &status, metadata, now)
            .await
        {
            Ok(outcome) => {
                if outcome.can_recover {
                    if let Some(previous) = self.cache.recover_to_healthy(wallet, now).await {
                        counter!("ht_liquidation_partial_exit_total").increment(1);
                        if let Some(updated_status) = self.cache.get_status(wallet).await {
                            self.emit_state_change(&previous, &updated_status, now);
                        }
                    }
                } else if let Some(current_status) = self.cache.get_status(wallet).await {
                    if let LiquidationState::PreLiquidation(current_metadata) = current_status.state
                    {
                        if current_metadata != outcome.metadata {
                            self.persist_pre_liquidation_metadata(wallet, outcome.metadata, now)
                                .await;
                        }
                    }
                }
            }
            Err(error) => {
                error!(
                    "Failed to run partial liquidation cycle for wallet {}: {}",
                    wallet, error
                );
            }
        }
    }

    // TODO: The reprice decision is driven by a wall-clock timer
    // (partial_reprice_interval_ms), not an engine command. For green/blue
    // deploys or multi-instance setups, this should become a
    // TickLiquidation engine command so all nodes agree on reprice timing
    // and only one instance submits orders. See also: NATS standby
    // controller could gate the watcher loop as a shorter-term fix.
    async fn run_partial_liquidation_cycle(
        &self,
        wallet: &WalletAddress,
        margin_mode: MarginMode,
        status: &AccountLiquidationStatus,
        mut metadata: PartialLiquidationMetadata,
        now: u64,
    ) -> anyhow::Result<PartialCycleOutcome> {
        metadata.mm_shortfall = status.shortfall();
        metadata.target_equity =
            target_equity_from_mm(status.mm_required, self.config.partial_target_buffer_bps);
        metadata.bonus_bps = compute_partial_bonus_bps(
            self.config.partial_bonus_start_bps,
            self.config.partial_bonus_max_bps,
            self.config.partial_bonus_ramp_ms,
            metadata.entered_at,
            now,
        );

        let above_target = is_above_partial_target(
            status.equity,
            status.mm_required,
            self.config.partial_target_buffer_bps,
        );
        let reprice_due = metadata.last_reprice_at.is_none_or(|last_reprice_at| {
            now.saturating_sub(last_reprice_at)
                >= self.config.partial_reprice_interval.as_millis() as u64
        });

        let sync_result = self
            .sync_and_cancel_orders(
                wallet,
                &metadata.active_order_client_ids,
                above_target || reprice_due,
            )
            .await?;

        if above_target {
            metadata.active_order_request_ids.clear();
            metadata.active_order_client_ids = sync_result.open_liquidation_client_ids;
            let can_recover = metadata.active_order_client_ids.is_empty();
            return Ok(PartialCycleOutcome {
                metadata,
                can_recover,
            });
        }

        metadata.active_order_client_ids = sync_result.open_liquidation_client_ids;
        if !reprice_due {
            if metadata.active_order_client_ids.is_empty() {
                metadata.active_order_request_ids.clear();
            }
            return Ok(PartialCycleOutcome {
                metadata,
                can_recover: false,
            });
        }

        let plans = match margin_mode {
            MarginMode::Standard => {
                self.build_standard_partial_orders(wallet, metadata.bonus_bps)
                    .await?
            }
            MarginMode::Portfolio => {
                self.build_portfolio_partial_orders(wallet, metadata.bonus_bps)
                    .await?
            }
        };

        metadata.last_reprice_at = Some(now);
        metadata.active_order_request_ids.clear();
        metadata.active_order_client_ids.clear();

        if plans.is_empty() {
            return Ok(PartialCycleOutcome {
                metadata,
                can_recover: false,
            });
        }

        let mut submitted_orders = 0u64;
        for (index, plan) in plans.iter().enumerate() {
            match self
                .submit_partial_liquidation_order(wallet, plan, now, index)
                .await
            {
                Ok((request_id, client_id, response)) => {
                    if response.status != OrderUpdateStatus::Rejected {
                        metadata.active_order_request_ids.push(request_id);
                        metadata.active_order_client_ids.push(client_id);
                        submitted_orders += 1;
                    } else {
                        warn!(
                            "Partial liquidation order rejected for wallet {} symbol {}: {}",
                            wallet,
                            plan.symbol,
                            response
                                .reason
                                .as_deref()
                                .unwrap_or("missing rejection reason")
                        );
                    }
                }
                Err(error) => {
                    error!(
                        "Failed to submit partial liquidation order for wallet {} symbol {}: {}",
                        wallet, plan.symbol, error
                    );
                }
            }
        }

        if submitted_orders > 0 {
            counter!("ht_liquidation_partial_orders_submitted_total").increment(submitted_orders);
        }

        Ok(PartialCycleOutcome {
            metadata,
            can_recover: false,
        })
    }

    async fn persist_pre_liquidation_metadata(
        &self,
        wallet: &WalletAddress,
        metadata: PartialLiquidationMetadata,
        now: u64,
    ) {
        let Some(current_status) = self.cache.get_status(wallet).await else {
            return;
        };

        if !matches!(current_status.state, LiquidationState::PreLiquidation(..)) {
            return;
        }

        let previous_state = current_status.state.clone();
        let mut updated_status = current_status;
        updated_status.state = LiquidationState::PreLiquidation(metadata);
        updated_status.updated_at = now;
        self.cache.set_status(updated_status.clone()).await;
        self.emit_state_change(&previous_state, &updated_status, now);
    }

    async fn sync_and_cancel_orders(
        &self,
        wallet: &WalletAddress,
        tracked_liquidation_client_ids: &[String],
        cancel_tracked_liquidation_orders: bool,
    ) -> anyhow::Result<OpenOrderSyncResult> {
        let order_snapshot = self
            .order_snapshot
            .as_ref()
            .context("OrderSnapshotProvider not set")?;
        let open_orders = order_snapshot.get_open_orders_for_wallet(wallet);
        let tracked_liquidation_client_ids: HashSet<&str> = tracked_liquidation_client_ids
            .iter()
            .map(String::as_str)
            .collect();
        let open_liquidation_client_ids = open_orders
            .iter()
            .filter_map(|order| order.client_id.as_deref())
            .filter(|client_id| tracked_liquidation_client_ids.contains(client_id))
            .map(str::to_owned)
            .collect::<Vec<_>>();
        let Some(current_positions) = self.try_current_position_sizes(wallet).await else {
            warn!(
                wallet = %wallet,
                "Skipping open-order liquidation cancel sync because portfolio snapshot is unavailable"
            );
            return Ok(OpenOrderSyncResult {
                open_liquidation_client_ids,
            });
        };
        let reduce_only_order_ids =
            self.reduce_only_open_order_ids(&current_positions, &open_orders);

        let mut cancel_targets = Vec::new();
        let mut seen_targets = HashSet::new();

        for order in &open_orders {
            let tracked_client_id = order
                .client_id
                .as_deref()
                .filter(|client_id| tracked_liquidation_client_ids.contains(client_id))
                .map(ToOwned::to_owned);

            let should_cancel = !reduce_only_order_ids.contains(&order.order_id)
                || (cancel_tracked_liquidation_orders && tracked_client_id.is_some());
            if !should_cancel {
                continue;
            }

            let target_key = tracked_client_id
                .clone()
                .unwrap_or_else(|| format!("oid:{}", order.order_id));
            if seen_targets.insert(target_key) {
                cancel_targets.push(CancelTarget {
                    order_id: order.order_id,
                    client_id: tracked_client_id.or_else(|| order.client_id.clone()),
                });
            }
        }

        for target in cancel_targets {
            if let Err(error) = self.cancel_open_order(wallet, &target).await {
                warn!(
                    "Failed to cancel open order {} for wallet {}: {}",
                    target.order_id, wallet, error
                );
            }
        }

        Ok(OpenOrderSyncResult {
            open_liquidation_client_ids,
        })
    }

    async fn build_standard_partial_orders(
        &self,
        wallet: &WalletAddress,
        bonus_bps: u32,
    ) -> anyhow::Result<Vec<PartialLiquidationOrderPlan>> {
        let builder = self
            .standard_account_builder
            .as_ref()
            .context("StandardAccountBuilder not set")?;
        let expected_prices = self.partial_execution_prices(wallet, bonus_bps).await?;
        let mut simulated_account = builder.build(wallet).await.map_err(|error| {
            anyhow!("Failed to build standard account for {}: {}", wallet, error)
        })?;
        let mut aggregated_plans = Vec::new();

        for _ in 0..PARTIAL_PLAN_MAX_ROUNDS {
            let margin = self
                .standard_margin_service
                .compute_margin(&simulated_account);
            let required_relief_amount = required_mm_relief(
                margin.equity,
                margin.position_mm,
                self.config.partial_target_buffer_bps,
            );
            if required_relief_amount <= PARTIAL_LIQ_MIN_SIZE {
                break;
            }

            let candidates = self.standard_partial_candidates(&simulated_account, &expected_prices);
            let round_plans = greedy_partial_plan(&candidates, required_relief_amount)
                .into_iter()
                .filter(|plan| plan.close_size > PARTIAL_LIQ_MIN_SIZE)
                .collect::<Vec<_>>();
            if round_plans.is_empty() {
                break;
            }

            for plan in round_plans {
                self.apply_standard_close(
                    &mut simulated_account,
                    &plan.symbol,
                    plan.close_size,
                    plan.limit_price,
                )?;
                merge_partial_plan(&mut aggregated_plans, plan);
            }
        }

        Ok(aggregated_plans)
    }

    async fn build_portfolio_partial_orders(
        &self,
        wallet: &WalletAddress,
        bonus_bps: u32,
    ) -> anyhow::Result<Vec<PartialLiquidationOrderPlan>> {
        let builder = self
            .risk_account_builder
            .as_ref()
            .context("RiskAccountBuilder not set")?;
        let expected_prices = self.partial_execution_prices(wallet, bonus_bps).await?;
        let mut simulated_snapshot = builder
            .build_snapshot(wallet)
            .await
            .map_err(|error| anyhow!("Failed to build PM snapshot for {}: {:?}", wallet, error))?
            .without_open_orders();
        let mut aggregated_plans = Vec::new();

        for _ in 0..PARTIAL_PLAN_MAX_ROUNDS {
            let margin = self.compute_pm_margin(builder, &simulated_snapshot)?;
            let required_relief_amount = required_mm_relief(
                margin.equity,
                margin.maintenance_margin_required,
                self.config.partial_target_buffer_bps,
            );
            if required_relief_amount <= PARTIAL_LIQ_MIN_SIZE {
                break;
            }

            let candidates =
                self.pm_partial_candidates(builder, &simulated_snapshot, &expected_prices)?;
            let round_plans = greedy_partial_plan(&candidates, required_relief_amount)
                .into_iter()
                .filter(|plan| plan.close_size > PARTIAL_LIQ_MIN_SIZE)
                .collect::<Vec<_>>();
            if round_plans.is_empty() {
                break;
            }

            for plan in round_plans {
                self.apply_pm_option_close(
                    &mut simulated_snapshot,
                    &plan.symbol,
                    plan.close_size,
                    plan.limit_price,
                )?;
                merge_partial_plan(&mut aggregated_plans, plan);
            }
        }

        Ok(aggregated_plans)
    }

    fn standard_partial_candidates(
        &self,
        account: &StandardAccount,
        expected_prices: &HashMap<String, Decimal>,
    ) -> Vec<LiquidationSliceCandidate> {
        let current_margin = self.standard_margin_service.compute_margin(account);
        let mut candidates = Vec::new();

        for option_position in account.short_options() {
            let Some(expected_price) = expected_prices.get(&option_position.symbol).copied() else {
                continue;
            };
            let close_step = option_position.abs_size().min(Decimal::ONE);
            if close_step <= PARTIAL_LIQ_MIN_SIZE {
                continue;
            }

            let mut simulated_account = account.clone();
            if self
                .apply_standard_close(
                    &mut simulated_account,
                    &option_position.symbol,
                    close_step,
                    expected_price,
                )
                .is_err()
            {
                continue;
            }

            let new_margin = self
                .standard_margin_service
                .compute_margin(&simulated_account);
            let mm_relief_per_unit =
                (current_margin.position_mm - new_margin.position_mm) / close_step;
            if mm_relief_per_unit <= Decimal::ZERO {
                continue;
            }

            candidates.push(LiquidationSliceCandidate {
                symbol: option_position.symbol.clone(),
                max_close_size: option_position.abs_size(),
                expected_fill_price: expected_price,
                mm_relief_per_unit,
            });
        }

        candidates
    }

    fn pm_partial_candidates(
        &self,
        builder: &crate::rsm::portfolio_margin::RiskAccountBuilder,
        snapshot: &PortfolioMarginSnapshot,
        expected_prices: &HashMap<String, Decimal>,
    ) -> anyhow::Result<Vec<LiquidationSliceCandidate>> {
        let current_margin = self.compute_pm_margin(builder, snapshot)?;
        let mut candidates = Vec::new();

        for underlying in &snapshot.underlyings {
            for option in &underlying.executed_options {
                if option.quantity >= Decimal::ZERO {
                    continue;
                }

                let symbol = format_pm_option_symbol(&option.key);
                let Some(expected_price) = expected_prices.get(&symbol).copied() else {
                    continue;
                };

                let close_step = option.quantity.abs().min(Decimal::ONE);
                if close_step <= PARTIAL_LIQ_MIN_SIZE {
                    continue;
                }

                let mut simulated_snapshot = snapshot.clone();
                if self
                    .apply_pm_option_close(
                        &mut simulated_snapshot,
                        &symbol,
                        close_step,
                        expected_price,
                    )
                    .is_err()
                {
                    continue;
                }

                let new_margin = self.compute_pm_margin(builder, &simulated_snapshot)?;
                let mm_relief_per_unit = (current_margin.maintenance_margin_required
                    - new_margin.maintenance_margin_required)
                    / close_step;
                if mm_relief_per_unit <= Decimal::ZERO {
                    continue;
                }

                candidates.push(LiquidationSliceCandidate {
                    symbol,
                    max_close_size: option.quantity.abs(),
                    expected_fill_price: expected_price,
                    mm_relief_per_unit,
                });
            }
        }

        Ok(candidates)
    }

    fn compute_pm_margin(
        &self,
        builder: &crate::rsm::portfolio_margin::RiskAccountBuilder,
        snapshot: &PortfolioMarginSnapshot,
    ) -> anyhow::Result<crate::types::MarginDetails> {
        let market_state = builder
            .resolve_market_state(snapshot, self.span_margin_service.config())
            .map_err(|error| anyhow!("Failed to resolve PM market state: {:?}", error))?;

        self.span_margin_service
            .compute_margin_from_snapshot(snapshot, market_state)
            .map_err(|error| anyhow!("Failed to compute PM margin: {}", error))
    }

    fn apply_standard_close(
        &self,
        account: &mut StandardAccount,
        symbol: &str,
        close_size: Decimal,
        execution_price: Decimal,
    ) -> anyhow::Result<()> {
        if close_size <= PARTIAL_LIQ_MIN_SIZE {
            return Ok(());
        }

        let option_index = account
            .option_positions
            .iter()
            .position(|position| position.symbol == symbol)
            .ok_or_else(|| anyhow!("Missing standard option position for {}", symbol))?;
        let current_size = account.option_positions[option_index].size;
        if current_size >= Decimal::ZERO {
            return Err(anyhow!(
                "Expected short standard option position for {}, found {}",
                symbol,
                current_size
            ));
        }
        if close_size > current_size.abs() + PARTIAL_LIQ_MIN_SIZE {
            return Err(anyhow!(
                "Close size {} exceeds standard short position {} for {}",
                close_size,
                current_size.abs(),
                symbol
            ));
        }

        account.usdc_balance -= execution_price * close_size;
        account.option_positions[option_index].size += close_size;
        if account.option_positions[option_index].size.abs() <= PARTIAL_LIQ_MIN_SIZE {
            account.option_positions.remove(option_index);
        }

        Ok(())
    }

    fn apply_pm_option_close(
        &self,
        snapshot: &mut PortfolioMarginSnapshot,
        symbol: &str,
        close_size: Decimal,
        execution_price: Decimal,
    ) -> anyhow::Result<()> {
        if close_size <= PARTIAL_LIQ_MIN_SIZE {
            return Ok(());
        }

        let parsed = ParsedSymbol::from_symbol(symbol)
            .map_err(|error| anyhow!("Failed to parse PM option symbol {}: {}", symbol, error))?;
        let option_key = PortfolioMarginOptionKey {
            underlying: parsed.underlying.clone(),
            option_type: parsed.option_type,
            strike: parsed.strike,
            expiry_ts: hypercall_types::expiry_date_to_timestamp(&parsed.underlying, parsed.expiry),
        };

        let underlying_snapshot = snapshot
            .underlyings
            .iter_mut()
            .find(|underlying| underlying.underlying == option_key.underlying)
            .ok_or_else(|| anyhow!("Missing PM underlying snapshot for {}", symbol))?;

        let option_exposure = underlying_snapshot
            .executed_options
            .iter_mut()
            .find(|option| option.key == option_key)
            .ok_or_else(|| anyhow!("Missing PM executed option exposure for {}", symbol))?;
        if option_exposure.quantity >= Decimal::ZERO {
            return Err(anyhow!(
                "Expected short PM option exposure for {}, found {}",
                symbol,
                option_exposure.quantity
            ));
        }
        if close_size > option_exposure.quantity.abs() + PARTIAL_LIQ_MIN_SIZE {
            return Err(anyhow!(
                "Close size {} exceeds PM short position {} for {}",
                close_size,
                option_exposure.quantity.abs(),
                symbol
            ));
        }

        let signed_closed_quantity = -close_size;
        snapshot.cash_balance +=
            signed_closed_quantity * (execution_price - option_exposure.entry_price);
        option_exposure.quantity += close_size;

        underlying_snapshot
            .executed_options
            .retain(|option| option.quantity.abs() > PARTIAL_LIQ_MIN_SIZE);
        snapshot.underlyings.retain(|underlying| {
            !underlying.executed_options.is_empty()
                || !underlying.executed_perps.is_empty()
                || !underlying.hypothetical_open_order_options.is_empty()
                || !underlying.hypothetical_open_order_perps.is_empty()
        });

        Ok(())
    }

    async fn partial_execution_prices(
        &self,
        wallet: &WalletAddress,
        bonus_bps: u32,
    ) -> anyhow::Result<HashMap<String, Decimal>> {
        let positions = self.current_positions(wallet).await;
        let mut prices = HashMap::new();

        for (symbol, position) in positions {
            if position.amount >= Decimal::ZERO {
                continue;
            }
            if ParsedSymbol::from_symbol(&symbol).is_err() {
                continue;
            }

            let fallback_mark_price = derive_portfolio_mark_price(&position)
                .with_context(|| format!("Failed to derive fallback mark for {}", symbol))?;
            let reference_price = self
                .quote_reference_price(&symbol)
                .unwrap_or(fallback_mark_price);
            if reference_price <= Decimal::ZERO {
                return Err(anyhow!(
                    "Expected positive liquidation reference price for {}, got {}",
                    symbol,
                    reference_price
                ));
            }

            prices.insert(
                symbol,
                reference_price * (Decimal::ONE + (Decimal::from(bonus_bps) / dec!(10000))),
            );
        }

        Ok(prices)
    }

    fn quote_reference_price(&self, symbol: &str) -> Option<Decimal> {
        // 1. Try BBO from orderbook
        if let Some(quote_provider) = self.quote_provider.as_ref() {
            if let Some(quote) = quote_provider.get_quote(symbol) {
                let bbo_price = quote
                    .best_ask
                    .filter(|price| *price > 0.0)
                    .or_else(|| quote.mid.filter(|price| *price > 0.0))
                    .or_else(|| quote.best_bid.filter(|price| *price > 0.0));
                if let Some(price) = bbo_price {
                    return Some(decimal_from_f64(price, "partial_quote_bbo"));
                }
            }
        }

        // 2. Fall back to theoretical price from vol surface
        if let Some(greeks_cache) = &self.greeks_cache {
            if let Ok(theo) =
                futures::executor::block_on(greeks_cache.get_theoretical_price(symbol))
            {
                if theo > 0.0 && theo.is_finite() {
                    return Some(decimal_from_f64(theo, "partial_quote_theo"));
                }
            }
        }

        None
    }

    async fn current_positions(&self, wallet: &WalletAddress) -> HashMap<String, PositionData> {
        self.portfolio_service
            .get_portfolio_balance(wallet)
            .await
            .unwrap_or_default()
            .positions
    }

    async fn try_current_position_sizes(
        &self,
        wallet: &WalletAddress,
    ) -> Option<HashMap<String, Decimal>> {
        self.portfolio_service
            .get_portfolio_balance(wallet)
            .await
            .map(|balance| {
                balance
                    .positions
                    .into_iter()
                    .map(|(symbol, position)| (symbol, position.amount))
                    .collect()
            })
    }

    fn is_order_reduce_only(
        &self,
        current_positions: &HashMap<String, Decimal>,
        order: &RuntimeOrderSummary,
    ) -> bool {
        let current_position = current_positions
            .get(&order.symbol)
            .copied()
            .unwrap_or(Decimal::ZERO);
        let signed_order_size = if matches!(order.side, Side::Buy) {
            order.remaining_size
        } else {
            -order.remaining_size
        };
        let new_position = current_position + signed_order_size;
        LiquidationManager::is_position_reduce_only(current_position, new_position)
    }

    fn reduce_only_open_order_ids(
        &self,
        current_positions: &HashMap<String, Decimal>,
        open_orders: &[RuntimeOrderSummary],
    ) -> HashSet<u64> {
        let mut remaining_positions = current_positions.clone();
        let mut sorted_orders: Vec<&RuntimeOrderSummary> = open_orders.iter().collect();
        sorted_orders.sort_by_key(|order| (order.created_at, order.order_id));

        let mut reduce_only_order_ids = HashSet::new();
        for order in sorted_orders {
            if !self.is_order_reduce_only(&remaining_positions, order) {
                continue;
            }

            let current_position = remaining_positions
                .get(&order.symbol)
                .copied()
                .unwrap_or(Decimal::ZERO);
            let signed_order_size = if matches!(order.side, Side::Buy) {
                order.remaining_size
            } else {
                -order.remaining_size
            };
            remaining_positions.insert(order.symbol.clone(), current_position + signed_order_size);
            reduce_only_order_ids.insert(order.order_id);
        }

        reduce_only_order_ids
    }

    async fn cancel_open_order(
        &self,
        wallet: &WalletAddress,
        target: &CancelTarget,
    ) -> anyhow::Result<()> {
        let request_id = Uuid::now_v7().to_string();
        let order_info = OrderInfo {
            symbol: String::new(),
            price: Decimal::ZERO,
            size: Decimal::ZERO,
            side: Side::Buy,
            tif: TimeInForce::GTC,
            client_id: target.client_id.clone(),
            order_id: Some(target.order_id),
            is_perp: false,
            underlying: None,
            reduce_only: None,
            nonce: None,
            signature: None,
            mmp_enabled: false,
            builder_code_address: None,
        };

        let response = self
            .send_engine_order(OrderActionMessage {
                timestamp: get_timestamp_millis(),
                info: order_info,
                action: OrderAction::CancelOrder,
                wallet: *wallet,
                api_wallet_address: Some(*wallet),
                mmp_triggered: false,
                request_id: Some(request_id),
            })
            .await?;

        let is_not_found = response.status == OrderUpdateStatus::Rejected
            && response.reason.as_deref().is_some_and(|reason| {
                reason.contains("already been completed")
                    || reason.contains("already filled")
                    || reason.contains("already cancelled")
                    || reason.contains("already canceled")
                    || reason.contains("not found")
            });

        if response.status == OrderUpdateStatus::Canceled || is_not_found {
            return Ok(());
        }

        Err(anyhow!(
            "Cancel rejected for wallet {} order {}: {}",
            wallet,
            target.order_id,
            response
                .reason
                .as_deref()
                .unwrap_or("missing cancellation reason")
        ))
    }

    async fn submit_partial_liquidation_order(
        &self,
        wallet: &WalletAddress,
        plan: &PartialLiquidationOrderPlan,
        now: u64,
        order_index: usize,
    ) -> anyhow::Result<(String, String, OrderUpdateMessage)> {
        let request_id = Uuid::now_v7().to_string();
        let client_id = format!("liq-{}-{}-{}", wallet, now, order_index);

        let response = self
            .send_engine_order(OrderActionMessage {
                timestamp: get_timestamp_millis(),
                info: OrderInfo {
                    symbol: plan.symbol.clone(),
                    price: plan.limit_price,
                    size: to_contract_units_decimal(&plan.symbol, plan.close_size),
                    side: Side::Buy,
                    tif: TimeInForce::GTC,
                    client_id: Some(client_id.clone()),
                    order_id: None,
                    is_perp: false,
                    underlying: None,
                    reduce_only: Some(true),
                    nonce: None,
                    signature: None,
                    mmp_enabled: false,
                    builder_code_address: None,
                },
                action: OrderAction::CreateOrder,
                wallet: *wallet,
                api_wallet_address: Some(*wallet),
                mmp_triggered: false,
                request_id: Some(request_id.clone()),
            })
            .await?;

        Ok((request_id, client_id, response))
    }

    async fn send_engine_order(
        &self,
        message: OrderActionMessage,
    ) -> anyhow::Result<OrderUpdateMessage> {
        let order_sender = self
            .order_sender
            .as_ref()
            .context("order sender not configured")?;
        let (response_tx, mut response_rx) = mpsc::channel(1);
        let request = UnifiedEngineRequest {
            message,
            response_tx,
            enqueued_at: Instant::now(),
            #[cfg(feature = "otel-tracing")]
            trace_context: None,
        };

        increment_pending_requests();
        order_sender
            .send(request)
            .await
            .context("failed to send liquidation order to engine")?;

        match timeout(ENGINE_RESPONSE_TIMEOUT, response_rx.recv()).await {
            Ok(Some(response)) => Ok(response),
            Ok(None) => Err(anyhow!("engine closed liquidation response channel")),
            Err(_) => Err(anyhow!("timed out waiting for liquidation engine response")),
        }
    }

    /// Emit a state change event.
    fn emit_state_change(
        &self,
        previous: &LiquidationState,
        status: &AccountLiquidationStatus,
        timestamp: u64,
    ) {
        info!(
            "Liquidation state change: wallet={}, {} -> {}",
            status.wallet,
            previous.as_str(),
            status.state.as_str()
        );

        // Publish to event bus via event sender
        if let Some(sender) = &self.event_sender {
            let msg = LiquidationStateMessage {
                wallet: status.wallet,
                previous_state: liquidation_state_to_type(previous),
                new_state: liquidation_state_to_type(&status.state),
                previous_liquidation_mode: previous
                    .liquidation_mode()
                    .map(|mode| mode.as_str().to_string()),
                liquidation_mode: status
                    .liquidation_mode()
                    .map(|mode| mode.as_str().to_string()),
                margin_mode: status.margin_mode.as_str().to_string(),
                equity: status.equity,
                mm_required: status.mm_required,
                maintenance_margin: status.maintenance_margin,
                shortfall: status.shortfall(),
                previous_auction_id: previous.auction_id().map(ToOwned::to_owned),
                projection_changed: has_material_projection_change(previous, &status.state),
                auction_id: status
                    .state
                    .auction_id()
                    .map(ToOwned::to_owned)
                    .or_else(|| previous.auction_id().map(ToOwned::to_owned)),
                status: status.clone(),
                timestamp,
            };

            if let Err(error) = sender.send(EngineMessage::LiquidationStateChange(msg)) {
                warn!("Failed to send liquidation state change event: {}", error);
            }
        }
    }
}

/// Get current timestamp in milliseconds.
fn get_timestamp_millis() -> u64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .unwrap_or_default()
        .as_millis() as u64
}

/// Convert LiquidationState to LiquidationStateType for messages.
fn liquidation_state_to_type(state: &LiquidationState) -> LiquidationStateType {
    match state {
        LiquidationState::Healthy => LiquidationStateType::Healthy,
        LiquidationState::PreLiquidation(..) => LiquidationStateType::PreLiquidation,
        LiquidationState::InLiquidation(..) => LiquidationStateType::InLiquidation,
        LiquidationState::Liquidated(..) => LiquidationStateType::Liquidated,
    }
}

fn decimal_from_f64(value: f64, field: &str) -> Decimal {
    if !value.is_finite() {
        warn!(
            "Liquidation {} value {} is not finite, clamping to zero",
            field, value
        );
        return Decimal::ZERO;
    }
    Decimal::from_f64_retain(value).unwrap_or_else(|| {
        warn!(
            "Liquidation {} value {} is not representable as Decimal, clamping to zero",
            field, value
        );
        Decimal::ZERO
    })
}

fn derive_portfolio_mark_price(position: &PositionData) -> anyhow::Result<Decimal> {
    if position.amount.abs() <= PARTIAL_LIQ_MIN_SIZE {
        return Err(anyhow!(
            "position amount {} is too small to derive mark price for {}",
            position.amount,
            position.symbol
        ));
    }

    Ok(position.entry_price + (position.unrealized_pnl / position.amount))
}

fn merge_partial_plan(
    aggregated_plans: &mut Vec<PartialLiquidationOrderPlan>,
    plan: PartialLiquidationOrderPlan,
) {
    if let Some(existing_plan) = aggregated_plans
        .iter_mut()
        .find(|existing| existing.symbol == plan.symbol && existing.limit_price == plan.limit_price)
    {
        existing_plan.close_size += plan.close_size;
        existing_plan.expected_mm_relief += plan.expected_mm_relief;
        return;
    }

    aggregated_plans.push(plan);
}

fn format_pm_option_symbol(key: &PortfolioMarginOptionKey) -> String {
    format!(
        "{}-{}-{}-{}",
        key.underlying,
        chrono::DateTime::from_timestamp(key.expiry_ts, 0)
            .expect("option expiry_ts must be valid")
            .format("%Y%m%d"),
        key.strike.normalize(),
        match key.option_type {
            crate::types::OptionType::Call => "C",
            crate::types::OptionType::Put => "P",
        }
    )
}

impl LiquidationWatcher {
    async fn has_short_option_positions(&self, wallet: &WalletAddress) -> Option<bool> {
        let Some(balance) = self.portfolio_service.get_portfolio_balance(wallet).await else {
            return None;
        };

        Some(balance.positions.iter().any(|(symbol, position)| {
            position.amount < Decimal::ZERO && ParsedSymbol::from_symbol(symbol).is_ok()
        }))
    }
}

#[async_trait::async_trait]
impl crate::shared::service::Service for LiquidationWatcher {
    fn name(&self) -> &'static str {
        "LiquidationWatcher"
    }

    fn owner(&self) -> crate::shared::service::ServiceOwner {
        crate::shared::service::ServiceOwner::Engine
    }

    async fn run(
        self: std::sync::Arc<Self>,
        shutdown: crate::shared::ShutdownRx,
    ) -> anyhow::Result<()> {
        self.run_with_shutdown(shutdown).await;
        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::portfolio::{PortfolioBalance, PortfolioChange, PortfolioError};
    use crate::rsm::ledger::{InMemoryLedger, Ledger};
    use crate::rsm::portfolio_margin::risk_account_builder::SpotPriceSource;
    use crate::types::{Config, Scenario, ScenarioType};
    use async_trait::async_trait;
    use hypercall_engine::FeeConfig;
    use hypercall_runtime_api::RuntimeOrderSummary;
    use hypercall_types::api_models::Portfolio;
    use hypercall_types::{to_human_readable_decimal, wallet_address::test_wallet};
    use std::any::Any;
    use std::collections::HashMap;
    use std::sync::atomic::{AtomicU64, Ordering};
    use std::sync::RwLock;
    use tokio::sync::RwLock as AsyncRwLock;
    use tokio::task::JoinHandle;

    #[derive(Default)]
    struct MockPortfolioService {
        balances: AsyncRwLock<HashMap<WalletAddress, PortfolioBalance>>,
    }

    impl MockPortfolioService {
        async fn set_balance(&self, wallet: WalletAddress, balance: PortfolioBalance) {
            self.balances.write().await.insert(wallet, balance);
        }

        async fn clear_balance(&self, wallet: &WalletAddress) {
            self.balances.write().await.remove(wallet);
        }
    }

    #[async_trait]
    impl PortfolioService for MockPortfolioService {
        async fn get_portfolio(&self, account: &WalletAddress) -> Portfolio {
            let balance = self
                .get_portfolio_balance(account)
                .await
                .unwrap_or_default();
            Portfolio {
                wallet_address: *account,
                positions: balance
                    .positions
                    .values()
                    .cloned()
                    .map(
                        |position| hypercall_types::api_models::PositionWithMetrics {
                            position: hypercall_types::api_models::Position {
                                wallet_address: *account,
                                symbol: position.symbol,
                                amount: position.amount,
                                entry_price: position.entry_price,
                                margin_posted: position.margin_posted,
                                realized_pnl: position.realized_pnl,
                                unrealized_pnl: position.unrealized_pnl,
                                updated_at: chrono::Utc::now(),
                            },
                            notional_value: Decimal::ZERO,
                            maintenance_margin: Decimal::ZERO,
                            liquidation_price: Decimal::ZERO,
                            margin_ratio: Decimal::ZERO,
                        },
                    )
                    .collect(),
                total_margin_used: balance.total_margin_used,
                available_balance: Decimal::ZERO,
                span_margin: None,
                margin_mode: "standard".to_string(),
                margin_summary: None,
            }
        }

        async fn get_portfolio_balance(&self, account: &WalletAddress) -> Option<PortfolioBalance> {
            self.balances.read().await.get(account).cloned()
        }

        async fn all_portfolios(&self) -> HashMap<WalletAddress, PortfolioBalance> {
            self.balances.read().await.clone()
        }

        async fn apply_event(
            &self,
            _event: &hypercall_types::EngineMessage,
        ) -> Result<Vec<PortfolioChange>, PortfolioError> {
            Ok(Vec::new())
        }

        async fn remove_expired_position(&self, _wallet: &WalletAddress, _symbol: &str) {}

        async fn apply_hypercore_position_update(
            &self,
            _update: &crate::portfolio::HypercorePositionUpdate,
        ) {
        }

        async fn set_hypercore_position(
            &self,
            _update: &crate::portfolio::HypercorePositionUpdate,
        ) {
        }

        fn as_any(&self) -> &dyn Any {
            self
        }

        async fn calculate_fill_accounting(
            &self,
            fill: &hypercall_types::Fill,
        ) -> Result<hypercall_types::FillAccounting, crate::portfolio::PortfolioError> {
            Ok(hypercall_types::FillAccounting::zero(fill.trade_id))
        }

        async fn apply_fill_to_memory(
            &self,
            _wallet: &WalletAddress,
            _symbol: &str,
            _side: &Side,
            _price: Decimal,
            _quantity: Decimal,
        ) {
        }
    }

    struct MockSpotPriceSource {
        prices: HashMap<String, f64>,
    }

    #[async_trait]
    impl SpotPriceSource for MockSpotPriceSource {
        async fn get_spot_price(&self, underlying: &str) -> Option<f64> {
            self.prices.get(underlying).copied()
        }
    }

    #[derive(Default)]
    struct MutableOrderSnapshotProvider {
        orders: RwLock<HashMap<WalletAddress, Vec<RuntimeOrderSummary>>>,
    }

    impl MutableOrderSnapshotProvider {
        fn insert_order(&self, wallet: WalletAddress, order: RuntimeOrderSummary) {
            self.orders
                .write()
                .expect("order lock poisoned")
                .entry(wallet)
                .or_default()
                .push(order);
        }

        fn remove_by_order_id(&self, wallet: &WalletAddress, order_id: u64) -> bool {
            let mut orders = self.orders.write().expect("order lock poisoned");
            let Some(wallet_orders) = orders.get_mut(wallet) else {
                return false;
            };
            let original_len = wallet_orders.len();
            wallet_orders.retain(|order| order.order_id != order_id);
            wallet_orders.len() != original_len
        }

        fn clear_wallet(&self, wallet: &WalletAddress) {
            self.orders
                .write()
                .expect("order lock poisoned")
                .remove(wallet);
        }
    }

    impl OrderSnapshotProvider for MutableOrderSnapshotProvider {
        fn get_open_orders_for_wallet(&self, wallet: &WalletAddress) -> Vec<RuntimeOrderSummary> {
            self.orders
                .read()
                .expect("order lock poisoned")
                .get(wallet)
                .cloned()
                .unwrap_or_default()
        }

        fn get_all_orders(&self) -> Vec<(RuntimeOrderSummary, WalletAddress)> {
            self.orders
                .read()
                .expect("order lock poisoned")
                .iter()
                .flat_map(|(wallet, orders)| {
                    orders.iter().cloned().map(move |order| (order, *wallet))
                })
                .collect()
        }
    }

    fn test_margin_config() -> Config {
        Config {
            risk_free_rate: 0.05,
            base_volatility: 0.8,
            base_skew: 0.0,
            base_excess_kurtosis: 0.0,
            scenarios: vec![
                Scenario {
                    scenario_type: ScenarioType::SpotChange,
                    value: 0.15,
                },
                Scenario {
                    scenario_type: ScenarioType::SpotChange,
                    value: -0.15,
                },
            ],
            delta_threshold: 0.0001,
            strike_match_tolerance: 0.01,
            expiry_match_tolerance_years: 0.001,
            allow_standard_margin_shorts: false,
            fee_config: FeeConfig::default(),
        }
    }

    fn liquidatable_health(wallet: &WalletAddress) -> LiquidationHealthResult {
        LiquidationHealthResult {
            wallet: wallet.to_string(),
            margin_mode: MarginMode::Standard,
            is_liquidatable: true,
            maintenance_margin: -200.0,
            equity: 100.0,
            mm_required: 300.0,
        }
    }

    fn recovered_health(wallet: &WalletAddress) -> LiquidationHealthResult {
        LiquidationHealthResult {
            wallet: wallet.to_string(),
            margin_mode: MarginMode::Standard,
            is_liquidatable: false,
            maintenance_margin: 120.0,
            equity: 420.0,
            mm_required: 300.0,
        }
    }

    struct StandardWatcherHarness {
        wallet: WalletAddress,
        symbol: String,
        cache: Arc<LiquidationCache>,
        portfolio_service: Arc<MockPortfolioService>,
        order_snapshot: Arc<MutableOrderSnapshotProvider>,
        watcher: LiquidationWatcher,
        engine_task: JoinHandle<()>,
    }

    async fn setup_standard_partial_liquidation_harness(
        reject_partial_orders: bool,
    ) -> StandardWatcherHarness {
        let wallet = test_wallet(42);
        let symbol = "BTC-20261231-100000-C".to_string();
        let cache = Arc::new(LiquidationCache::new());
        let portfolio_service = Arc::new(MockPortfolioService::default());
        portfolio_service
            .set_balance(
                wallet,
                PortfolioBalance {
                    positions: HashMap::from([(
                        symbol.clone(),
                        PositionData {
                            symbol: symbol.clone(),
                            amount: dec!(-4),
                            entry_price: dec!(100),
                            margin_posted: Decimal::ZERO,
                            realized_pnl: Decimal::ZERO,
                            unrealized_pnl: Decimal::ZERO,
                        },
                    )]),
                    total_margin_used: Decimal::ZERO,
                },
            )
            .await;

        let ledger = Arc::new(InMemoryLedger::new());
        ledger
            .set_balance(&wallet, dec!(100))
            .await
            .expect("ledger balance should be seeded");

        let standard_account_builder = Arc::new(StandardAccountBuilder::new(
            ledger,
            portfolio_service.clone(),
            Arc::new(MockSpotPriceSource {
                prices: HashMap::from([("BTC".to_string(), 105000.0)]),
            }),
        ));
        let quote_provider = Arc::new(crate::rsm::engine_snapshot::MockQuoteProvider::new());
        quote_provider.set_quote(
            &symbol,
            hypercall_runtime_api::SnapshotBookQuote {
                best_bid: Some(149.0),
                best_bid_size: Some(10.0),
                best_ask: Some(150.0),
                best_ask_size: Some(10.0),
                mid: Some(149.5),
                bids: vec![(149.0, 10.0)],
                asks: vec![(150.0, 10.0)],
            },
        );

        let order_snapshot = Arc::new(MutableOrderSnapshotProvider::default());
        order_snapshot.insert_order(
            wallet,
            RuntimeOrderSummary {
                order_id: 7,
                symbol: symbol.clone(),
                side: Side::Sell,
                price: dec!(95),
                original_size: dec!(1),
                remaining_size: dec!(1),
                is_perp: false,
                mmp_enabled: false,
                client_id: Some("risk-increasing".to_string()),
                created_at: 1,
            },
        );

        let (order_tx, mut order_rx) = mpsc::channel::<UnifiedEngineRequest>(8);
        let next_order_id = Arc::new(AtomicU64::new(1_000));
        let mock_orders = order_snapshot.clone();
        let order_id_counter = next_order_id.clone();
        let engine_task = tokio::spawn(async move {
            while let Some(request) = order_rx.recv().await {
                let message = request.message.clone();
                let response = match message.action {
                    OrderAction::CreateOrder if reject_partial_orders => OrderUpdateMessage {
                        timestamp: message.timestamp,
                        info: message.info,
                        status: OrderUpdateStatus::Rejected,
                        reason: Some("forced rejection".to_string()),
                        filled_size: Decimal::ZERO,
                        order_id: None,
                        wallet_address: message.wallet,
                        mmp_triggered: false,
                        request_id: message.request_id,
                    },
                    OrderAction::CreateOrder => {
                        let order_id = order_id_counter.fetch_add(1, Ordering::Relaxed);
                        let human_size =
                            to_human_readable_decimal(&message.info.symbol, message.info.size);
                        mock_orders.insert_order(
                            message.wallet,
                            RuntimeOrderSummary {
                                order_id,
                                symbol: message.info.symbol.clone(),
                                side: message.info.side,
                                price: message.info.price,
                                original_size: human_size,
                                remaining_size: human_size,
                                is_perp: message.info.is_perp,
                                mmp_enabled: message.info.mmp_enabled,
                                client_id: message.info.client_id.clone(),
                                created_at: message.timestamp as i64,
                            },
                        );
                        OrderUpdateMessage {
                            timestamp: message.timestamp,
                            info: message.info,
                            status: OrderUpdateStatus::Acked,
                            reason: None,
                            filled_size: Decimal::ZERO,
                            order_id: Some(order_id),
                            wallet_address: message.wallet,
                            mmp_triggered: false,
                            request_id: message.request_id,
                        }
                    }
                    OrderAction::CancelOrder => {
                        let removed = message
                            .info
                            .order_id
                            .map(|order_id| {
                                mock_orders.remove_by_order_id(&message.wallet, order_id)
                            })
                            .unwrap_or(false);
                        OrderUpdateMessage {
                            timestamp: message.timestamp,
                            info: message.info,
                            status: if removed {
                                OrderUpdateStatus::Canceled
                            } else {
                                OrderUpdateStatus::Rejected
                            },
                            reason: if removed {
                                None
                            } else {
                                Some("order not found".to_string())
                            },
                            filled_size: Decimal::ZERO,
                            order_id: None,
                            wallet_address: message.wallet,
                            mmp_triggered: false,
                            request_id: message.request_id,
                        }
                    }
                    OrderAction::ReplaceOrder => OrderUpdateMessage {
                        timestamp: message.timestamp,
                        info: message.info,
                        status: OrderUpdateStatus::Rejected,
                        reason: Some(
                            "replace orders unsupported in liquidation mock engine".to_string(),
                        ),
                        filled_size: Decimal::ZERO,
                        order_id: None,
                        wallet_address: message.wallet,
                        mmp_triggered: false,
                        request_id: message.request_id,
                    },
                };

                request
                    .response_tx
                    .send(response)
                    .await
                    .expect("mock engine should deliver response");
            }
        });

        let watcher = LiquidationWatcher::new_for_tests_without_tier_cache(
            cache.clone(),
            portfolio_service.clone(),
            Arc::new(SpanMarginService::new_for_tests(test_margin_config())),
            Arc::new(crate::standard_margin::StandardMarginService::new()),
            LiquidationWatcherConfig {
                poll_interval: Duration::from_secs(5),
                full_escalation_timeout: Duration::from_secs(60),
                min_shortfall_threshold: Decimal::ZERO,
                partial_target_buffer_bps: 500,
                partial_reprice_interval: Duration::from_secs(5),
                partial_bonus_start_bps: 50,
                partial_bonus_max_bps: 500,
                partial_bonus_ramp_ms: 60_000,
            },
        )
        .with_standard_account_builder(standard_account_builder)
        .with_quote_provider(quote_provider)
        .with_order_snapshot(order_snapshot.clone())
        .with_order_sender(order_tx);

        StandardWatcherHarness {
            wallet,
            symbol,
            cache,
            portfolio_service,
            order_snapshot,
            watcher,
            engine_task,
        }
    }

    #[test]
    fn test_config_default() {
        let config = LiquidationWatcherConfig::default();
        assert_eq!(config.poll_interval, Duration::from_secs(5));
        assert_eq!(config.full_escalation_timeout, Duration::from_secs(60));
        assert_eq!(config.min_shortfall_threshold, dec!(0));
        assert_eq!(config.partial_target_buffer_bps, 500);
        assert_eq!(config.partial_reprice_interval, Duration::from_secs(5));
        assert_eq!(config.partial_bonus_start_bps, 50);
        assert_eq!(config.partial_bonus_max_bps, 1_000);
        assert_eq!(config.partial_bonus_ramp_ms, 300_000);
    }

    #[test]
    fn test_config_from_runtime() {
        let runtime = crate::backend_config::LiquidationRuntimeConfig {
            enabled: true,
            health_poll_interval_ms: 1_500,
            partial_target_buffer_bps: 250,
            partial_reprice_interval_ms: 2_500,
            partial_bonus_start_bps: 75,
            partial_bonus_max_bps: 500,
            partial_bonus_ramp_ms: 120_000,
            min_shortfall_threshold: dec!(42),
            full_escalation_timeout_ms: 9_000,
            full_target_buffer_bps: 700,
            chain_observer_poll_interval_ms: 1_000,
            chain_observer_max_lag_blocks: 16,
        };

        let config = LiquidationWatcherConfig::from_runtime_config(&runtime);
        assert_eq!(config.poll_interval, Duration::from_millis(1_500));
        assert_eq!(config.full_escalation_timeout, Duration::from_millis(9_000));
        assert_eq!(config.min_shortfall_threshold, dec!(42));
        assert_eq!(config.partial_target_buffer_bps, 250);
        assert_eq!(
            config.partial_reprice_interval,
            Duration::from_millis(2_500)
        );
        assert_eq!(config.partial_bonus_start_bps, 75);
        assert_eq!(config.partial_bonus_max_bps, 500);
        assert_eq!(config.partial_bonus_ramp_ms, 120_000);
    }

    #[test]
    fn test_merge_partial_plan_accumulates_same_symbol_and_price() {
        let mut plans = vec![PartialLiquidationOrderPlan {
            symbol: "BTC-20251231-100000-C".to_string(),
            close_size: dec!(1),
            limit_price: dec!(100),
            expected_mm_relief: dec!(50),
        }];

        merge_partial_plan(
            &mut plans,
            PartialLiquidationOrderPlan {
                symbol: "BTC-20251231-100000-C".to_string(),
                close_size: dec!(2),
                limit_price: dec!(100),
                expected_mm_relief: dec!(75),
            },
        );

        assert_eq!(plans.len(), 1);
        assert_eq!(plans[0].close_size, dec!(3));
        assert_eq!(plans[0].expected_mm_relief, dec!(125));
    }

    #[test]
    fn test_derive_portfolio_mark_price() {
        let position = PositionData {
            symbol: "BTC-20251231-100000-C".to_string(),
            amount: dec!(-2),
            entry_price: dec!(100),
            margin_posted: Decimal::ZERO,
            realized_pnl: Decimal::ZERO,
            unrealized_pnl: dec!(20),
        };

        assert_eq!(derive_portfolio_mark_price(&position).unwrap(), dec!(90));
    }

    #[test]
    fn test_format_pm_option_symbol() {
        let symbol = format_pm_option_symbol(&PortfolioMarginOptionKey {
            underlying: "BTC".to_string(),
            option_type: crate::types::OptionType::Call,
            strike: dec!(100000),
            expiry_ts: hypercall_types::expiry_date_to_timestamp("BTC", 20251231),
        });

        assert_eq!(symbol, "BTC-20251231-100000-C");
    }

    #[tokio::test]
    async fn test_watcher_enters_partial_liquidation_and_recovers_after_orders_clear() {
        let harness = setup_standard_partial_liquidation_harness(false).await;

        harness
            .watcher
            .process_health_check(
                &harness.wallet,
                &liquidatable_health(&harness.wallet),
                MarginMode::Standard,
                10_000,
            )
            .await;

        let status = harness
            .cache
            .get_status(&harness.wallet)
            .await
            .expect("wallet should have liquidation state");
        let metadata = match status.state {
            LiquidationState::PreLiquidation(metadata) => metadata,
            other => panic!("expected pre-liquidation state, got {:?}", other),
        };
        assert_eq!(metadata.mm_shortfall, dec!(200));
        assert_eq!(metadata.target_equity, dec!(315.00));
        assert_eq!(metadata.active_order_client_ids.len(), 1);
        assert!(
            metadata
                .active_order_client_ids
                .iter()
                .all(|client_id| client_id.starts_with("liq-")),
            "partial liquidation orders should use liquidation client ids"
        );

        let open_orders = harness
            .order_snapshot
            .get_open_orders_for_wallet(&harness.wallet);
        assert_eq!(open_orders.len(), 1);
        assert_eq!(open_orders[0].side, Side::Buy);
        assert_eq!(open_orders[0].symbol, harness.symbol);
        assert!(open_orders[0]
            .client_id
            .as_deref()
            .is_some_and(|client_id| client_id.starts_with("liq-")));

        harness.order_snapshot.clear_wallet(&harness.wallet);

        harness
            .watcher
            .process_health_check(
                &harness.wallet,
                &recovered_health(&harness.wallet),
                MarginMode::Standard,
                20_000,
            )
            .await;

        let recovered_status = harness
            .cache
            .get_status(&harness.wallet)
            .await
            .expect("wallet should remain tracked after recovery");
        assert!(recovered_status.state.is_healthy());

        harness.engine_task.abort();
    }

    #[tokio::test]
    async fn test_watcher_requires_one_extra_cycle_to_recover_after_canceling_open_orders() {
        let harness = setup_standard_partial_liquidation_harness(false).await;

        harness
            .watcher
            .process_health_check(
                &harness.wallet,
                &liquidatable_health(&harness.wallet),
                MarginMode::Standard,
                10_000,
            )
            .await;

        harness
            .watcher
            .process_health_check(
                &harness.wallet,
                &recovered_health(&harness.wallet),
                MarginMode::Standard,
                20_000,
            )
            .await;

        let intermediate_status = harness
            .cache
            .get_status(&harness.wallet)
            .await
            .expect("wallet should remain tracked after cancel cycle");
        let intermediate_metadata = match intermediate_status.state {
            LiquidationState::PreLiquidation(metadata) => metadata,
            other => panic!(
                "expected pre-liquidation state after cancel cycle, got {:?}",
                other
            ),
        };
        assert_eq!(intermediate_metadata.active_order_client_ids.len(), 1);
        assert!(
            harness
                .order_snapshot
                .get_open_orders_for_wallet(&harness.wallet)
                .is_empty(),
            "tracked liquidation orders should have been canceled in the recovery cycle"
        );

        harness
            .watcher
            .process_health_check(
                &harness.wallet,
                &recovered_health(&harness.wallet),
                MarginMode::Standard,
                25_000,
            )
            .await;

        let recovered_status = harness
            .cache
            .get_status(&harness.wallet)
            .await
            .expect("wallet should remain tracked after recovery");
        assert!(recovered_status.state.is_healthy());

        harness.engine_task.abort();
    }

    #[tokio::test]
    async fn test_sync_and_cancel_orders_cancels_excess_reduce_only_orders() {
        let harness = setup_standard_partial_liquidation_harness(false).await;
        harness.order_snapshot.clear_wallet(&harness.wallet);
        harness
            .portfolio_service
            .set_balance(
                harness.wallet,
                PortfolioBalance {
                    positions: HashMap::from([(
                        harness.symbol.clone(),
                        PositionData {
                            symbol: harness.symbol.clone(),
                            amount: dec!(-1),
                            entry_price: dec!(100),
                            margin_posted: Decimal::ZERO,
                            realized_pnl: Decimal::ZERO,
                            unrealized_pnl: Decimal::ZERO,
                        },
                    )]),
                    total_margin_used: Decimal::ZERO,
                },
            )
            .await;
        harness.order_snapshot.insert_order(
            harness.wallet,
            RuntimeOrderSummary {
                order_id: 100,
                symbol: harness.symbol.clone(),
                side: Side::Buy,
                price: dec!(150),
                original_size: dec!(1),
                remaining_size: dec!(1),
                is_perp: false,
                mmp_enabled: false,
                client_id: Some("close-1".to_string()),
                created_at: 10,
            },
        );
        harness.order_snapshot.insert_order(
            harness.wallet,
            RuntimeOrderSummary {
                order_id: 101,
                symbol: harness.symbol.clone(),
                side: Side::Buy,
                price: dec!(151),
                original_size: dec!(1),
                remaining_size: dec!(1),
                is_perp: false,
                mmp_enabled: false,
                client_id: Some("close-2".to_string()),
                created_at: 20,
            },
        );

        harness
            .watcher
            .sync_and_cancel_orders(&harness.wallet, &[], false)
            .await
            .expect("sync should succeed");

        let open_orders = harness
            .order_snapshot
            .get_open_orders_for_wallet(&harness.wallet);
        assert_eq!(open_orders.len(), 1);
        assert_eq!(open_orders[0].order_id, 100);

        harness.engine_task.abort();
    }

    #[tokio::test]
    async fn test_sync_and_cancel_orders_skips_when_portfolio_snapshot_missing() {
        let harness = setup_standard_partial_liquidation_harness(false).await;
        harness.order_snapshot.clear_wallet(&harness.wallet);
        harness
            .portfolio_service
            .clear_balance(&harness.wallet)
            .await;
        harness.order_snapshot.insert_order(
            harness.wallet,
            RuntimeOrderSummary {
                order_id: 100,
                symbol: harness.symbol.clone(),
                side: Side::Buy,
                price: dec!(150),
                original_size: dec!(1),
                remaining_size: dec!(1),
                is_perp: false,
                mmp_enabled: false,
                client_id: Some("liq-keep".to_string()),
                created_at: 10,
            },
        );

        let sync_result = harness
            .watcher
            .sync_and_cancel_orders(&harness.wallet, &["liq-keep".to_string()], false)
            .await
            .expect("sync should succeed");

        assert_eq!(sync_result.open_liquidation_client_ids, vec!["liq-keep"]);
        let open_orders = harness
            .order_snapshot
            .get_open_orders_for_wallet(&harness.wallet);
        assert_eq!(open_orders.len(), 1);
        assert_eq!(open_orders[0].order_id, 100);

        harness.engine_task.abort();
    }

    #[tokio::test]
    async fn test_watcher_keeps_pre_liquidation_when_partial_order_is_rejected() {
        let harness = setup_standard_partial_liquidation_harness(true).await;

        harness
            .watcher
            .process_health_check(
                &harness.wallet,
                &liquidatable_health(&harness.wallet),
                MarginMode::Standard,
                10_000,
            )
            .await;

        let status = harness
            .cache
            .get_status(&harness.wallet)
            .await
            .expect("wallet should have liquidation state");
        let metadata = match status.state {
            LiquidationState::PreLiquidation(metadata) => metadata,
            other => panic!("expected pre-liquidation state, got {:?}", other),
        };

        assert!(
            metadata.active_order_request_ids.is_empty(),
            "rejected liquidation orders must not remain tracked as active requests"
        );
        assert!(
            metadata.active_order_client_ids.is_empty(),
            "rejected liquidation orders must not remain tracked as active client ids"
        );
        assert_eq!(metadata.last_reprice_at, Some(10_000));
        assert!(
            harness
                .order_snapshot
                .get_open_orders_for_wallet(&harness.wallet)
                .iter()
                .all(|order| !order
                    .client_id
                    .as_deref()
                    .is_some_and(|client_id| client_id.starts_with("liq-"))),
            "rejected liquidation orders must not appear in the open-order snapshot"
        );

        harness.engine_task.abort();
    }

    #[tokio::test]
    async fn test_watcher_continues_polling_tracked_wallets_after_positions_close() {
        let harness = setup_standard_partial_liquidation_harness(false).await;

        harness
            .watcher
            .process_health_check(
                &harness.wallet,
                &liquidatable_health(&harness.wallet),
                MarginMode::Standard,
                10_000,
            )
            .await;

        harness
            .portfolio_service
            .set_balance(
                harness.wallet,
                PortfolioBalance {
                    positions: HashMap::new(),
                    total_margin_used: Decimal::ZERO,
                },
            )
            .await;

        harness.watcher.check_all_accounts().await.unwrap();

        let intermediate_status = harness
            .cache
            .get_status(&harness.wallet)
            .await
            .expect("tracked wallet should still be processed after positions close");
        assert!(
            matches!(
                intermediate_status.state,
                LiquidationState::PreLiquidation(..)
            ),
            "first recovery cycle should cancel tracked liquidation orders before clearing state"
        );
        assert!(
            harness
                .order_snapshot
                .get_open_orders_for_wallet(&harness.wallet)
                .is_empty(),
            "tracked liquidation orders should be canceled even after positions close"
        );

        harness.watcher.check_all_accounts().await.unwrap();

        let recovered_status = harness
            .cache
            .get_status(&harness.wallet)
            .await
            .expect("wallet should remain tracked after recovery");
        assert!(recovered_status.state.is_healthy());

        harness.engine_task.abort();
    }
}