hypercall/observability/metrics_collector/

risk.rs

use super::*;
use crate::read_cache::portfolio::{PortfolioSummary, WalletMarginSnapshot};
use hypercall_types::WalletAddress;

struct RiskyAccount {
    wallet_short: String,
    margin_utilization: f64,
    equity: f64,
}

#[derive(Default)]
struct RiskMetricsSnapshot {
    total_equity: f64,
    total_margin_required: f64,
    accounts_near_liquidation: i64,
    accounts_in_breach: i64,
    accounts_negative_available: i64,
    accounts_safe: i64,
    accounts_warning: i64,
    accounts_danger: i64,
    accounts_critical: i64,
    equity_under_1k: i64,
    equity_1k_10k: i64,
    equity_10k_100k: i64,
    equity_100k_1m: i64,
    equity_over_1m: i64,
    max_position_notional: f64,
    max_account_equity: f64,
    position_conversion_failures: i64,
    risky_accounts: Vec<RiskyAccount>,
    margin_failures: i64,
}

impl RiskMetricsSnapshot {
    fn margin_utilization(&self) -> f64 {
        if self.total_equity > 0.0 {
            (self.total_margin_required / self.total_equity * 100.0).min(100.0)
        } else {
            0.0
        }
    }

    fn record_margin_snapshot(
        &mut self,
        wallet: &WalletAddress,
        snapshot: &WalletMarginSnapshot,
        near_liquidation_threshold: f64,
    ) {
        let Some(values) = RiskMarginValues::from_snapshot(snapshot) else {
            warn!(
                wallet = %wallet,
                "Margin decimal conversion failed, skipping wallet"
            );
            self.margin_failures += 1;
            return;
        };

        self.total_equity += values.equity;
        self.total_margin_required += values.maintenance_margin_required;
        self.max_account_equity = self.max_account_equity.max(values.equity);

        self.record_breach_state(wallet, &values);
        self.record_margin_utilization(wallet, &values, near_liquidation_threshold);
    }

    fn record_breach_state(&mut self, wallet: &WalletAddress, values: &RiskMarginValues) {
        if values.maintenance_margin_required > 0.0
            && values.equity < values.maintenance_margin_required
        {
            self.accounts_in_breach += 1;
            warn!(
                "ACCOUNT IN BREACH: {} has equity ${:.2} < MM ${:.2}",
                wallet, values.equity, values.maintenance_margin_required
            );
        }

        if values.available_balance < 0.0 {
            self.accounts_negative_available += 1;
            warn!(
                "NEGATIVE AVAILABLE: {} has available ${:.2} (equity ${:.2}, IM/reserved ${:.2}, margin excess ${:.2})",
                wallet,
                values.available_balance,
                values.equity,
                values.initial_margin_required,
                values.initial_margin_excess
            );
        }
    }

    fn record_margin_utilization(
        &mut self,
        wallet: &WalletAddress,
        values: &RiskMarginValues,
        near_liquidation_threshold: f64,
    ) {
        if values.equity <= 0.0 {
            return;
        }

        let margin_utilization =
            (values.maintenance_margin_required / values.equity * 100.0).min(100.0);
        self.record_risk_level(margin_utilization);
        self.record_equity_bucket(values.equity);

        if values.maintenance_margin_required > 0.0 {
            self.risky_accounts.push(RiskyAccount {
                wallet_short: short_wallet(wallet),
                margin_utilization,
                equity: values.equity,
            });

            let margin_buffer = (values.equity - values.maintenance_margin_required)
                / values.maintenance_margin_required;
            if margin_buffer < near_liquidation_threshold && margin_buffer > 0.0 {
                self.accounts_near_liquidation += 1;
            }
        }
    }

    fn record_risk_level(&mut self, margin_utilization: f64) {
        if margin_utilization < 50.0 {
            self.accounts_safe += 1;
        } else if margin_utilization < 80.0 {
            self.accounts_warning += 1;
        } else if margin_utilization < 95.0 {
            self.accounts_danger += 1;
        } else {
            self.accounts_critical += 1;
        }
    }

    fn record_equity_bucket(&mut self, equity: f64) {
        if equity < 1_000.0 {
            self.equity_under_1k += 1;
        } else if equity < 10_000.0 {
            self.equity_1k_10k += 1;
        } else if equity < 100_000.0 {
            self.equity_10k_100k += 1;
        } else if equity < 1_000_000.0 {
            self.equity_100k_1m += 1;
        } else {
            self.equity_over_1m += 1;
        }
    }

    fn record_positions(&mut self, portfolio: &PortfolioSummary) {
        for position in portfolio.positions.values() {
            let Some(amount) = position.amount.to_f64() else {
                warn!(
                    symbol = %position.symbol,
                    amount = %position.amount,
                    "Failed to convert position amount for risk exposure metrics"
                );
                self.position_conversion_failures += 1;
                continue;
            };
            let Some(entry_price) = position.entry_price.to_f64() else {
                warn!(
                    symbol = %position.symbol,
                    entry_price = %position.entry_price,
                    "Failed to convert entry price for risk exposure metrics"
                );
                self.position_conversion_failures += 1;
                continue;
            };
            let amount = amount.abs();
            self.max_position_notional = self.max_position_notional.max(amount * entry_price);
        }
    }

    fn top_risky_accounts(&mut self) -> Vec<RiskyAccount> {
        self.risky_accounts.sort_by(|a, b| {
            b.margin_utilization
                .partial_cmp(&a.margin_utilization)
                .unwrap_or(std::cmp::Ordering::Equal)
        });
        self.risky_accounts
            .drain(..self.risky_accounts.len().min(5))
            .collect()
    }
}

struct RiskMarginValues {
    equity: f64,
    maintenance_margin_required: f64,
    initial_margin_required: f64,
    available_balance: f64,
    initial_margin_excess: f64,
}

impl RiskMarginValues {
    fn from_snapshot(snapshot: &WalletMarginSnapshot) -> Option<Self> {
        Some(Self {
            equity: snapshot.margin_summary.equity.to_f64()?,
            maintenance_margin_required: snapshot
                .span_margin
                .maintenance_margin_required
                .to_f64()?,
            initial_margin_required: snapshot.total_margin_used.to_f64()?,
            available_balance: snapshot.available_balance.to_f64()?,
            initial_margin_excess: snapshot.margin_summary.initial_margin.to_f64()?,
        })
    }
}

impl MetricsCollector {
    // ===== Risk Metrics =====

    pub(super) async fn collect_risk_metrics(&self) {
        let portfolios = self.portfolio_cache.get_all_portfolios().await;
        let mut snapshot = RiskMetricsSnapshot::default();

        for (wallet, portfolio) in &portfolios {
            match self
                .portfolio_cache
                .compute_wallet_margin_snapshot(wallet)
                .await
            {
                Ok(margin_snapshot) => snapshot.record_margin_snapshot(
                    wallet,
                    &margin_snapshot,
                    self.config.near_liquidation_threshold,
                ),
                Err(e) => {
                    warn!(
                        wallet = %wallet,
                        error = %e,
                        "Failed to compute wallet margin snapshot"
                    );
                    snapshot.margin_failures += 1;
                }
            }

            snapshot.record_positions(portfolio);
        }

        emit_risk_metrics(&mut snapshot);
        log_risk_summary(&snapshot);
    }
}

fn emit_risk_metrics(snapshot: &mut RiskMetricsSnapshot) {
    let margin_utilization = snapshot.margin_utilization();

    gauge!("ht_margin_utilization_percent").set(margin_utilization);
    gauge!("ht_accounts_near_liquidation").set(snapshot.accounts_near_liquidation as f64);
    gauge!("ht_total_equity_usd").set(snapshot.total_equity);
    gauge!("ht_total_margin_required_usd").set(snapshot.total_margin_required);
    gauge!("ht_margin_calc_failures").set(snapshot.margin_failures as f64);
    gauge!("ht_risk_metrics_complete").set(if snapshot.margin_failures == 0 {
        1.0
    } else {
        0.0
    });

    gauge!("ht_accounts_in_breach").set(snapshot.accounts_in_breach as f64);
    gauge!("ht_accounts_negative_available").set(snapshot.accounts_negative_available as f64);

    gauge!("ht_accounts_by_risk_level", "level" => "safe").set(snapshot.accounts_safe as f64);
    gauge!("ht_accounts_by_risk_level", "level" => "warning").set(snapshot.accounts_warning as f64);
    gauge!("ht_accounts_by_risk_level", "level" => "danger").set(snapshot.accounts_danger as f64);
    gauge!("ht_accounts_by_risk_level", "level" => "critical")
        .set(snapshot.accounts_critical as f64);

    gauge!("ht_accounts_by_equity", "bucket" => "<1k").set(snapshot.equity_under_1k as f64);
    gauge!("ht_accounts_by_equity", "bucket" => "1k-10k").set(snapshot.equity_1k_10k as f64);
    gauge!("ht_accounts_by_equity", "bucket" => "10k-100k").set(snapshot.equity_10k_100k as f64);
    gauge!("ht_accounts_by_equity", "bucket" => "100k-1M").set(snapshot.equity_100k_1m as f64);
    gauge!("ht_accounts_by_equity", "bucket" => ">1M").set(snapshot.equity_over_1m as f64);

    gauge!("ht_max_position_notional_usd").set(snapshot.max_position_notional);
    gauge!("ht_max_account_equity_usd").set(snapshot.max_account_equity);
    gauge!("ht_risk_position_conversion_failures")
        .set(snapshot.position_conversion_failures as f64);

    let top_risky = snapshot.top_risky_accounts();
    emit_top_risky_accounts(&top_risky);
    log_danger_accounts(&top_risky);
}

fn emit_top_risky_accounts(top_risky: &[RiskyAccount]) {
    for (rank, account) in top_risky.iter().enumerate() {
        let rank_str = (rank + 1).to_string();
        gauge!(
            "ht_top_risky_account_margin_pct",
            "rank" => rank_str.clone(),
            "wallet" => account.wallet_short.clone()
        )
        .set(account.margin_utilization);
        gauge!(
            "ht_top_risky_account_equity",
            "rank" => rank_str,
            "wallet" => account.wallet_short.clone()
        )
        .set(account.equity);
    }
}

fn log_danger_accounts(top_risky: &[RiskyAccount]) {
    let danger_accounts: Vec<_> = top_risky
        .iter()
        .filter(|account| account.margin_utilization >= 80.0)
        .collect();
    if danger_accounts.is_empty() {
        return;
    }

    warn!(
        "⚠️ {} accounts in danger zone: {:?}",
        danger_accounts.len(),
        danger_accounts
            .iter()
            .map(|account| format!(
                "{}: {:.1}%",
                account.wallet_short, account.margin_utilization
            ))
            .collect::<Vec<_>>()
    );
}

fn log_risk_summary(snapshot: &RiskMetricsSnapshot) {
    if snapshot.margin_failures > 0 {
        warn!(
            "Risk metrics omitted {} accounts because margin calculation failed; aggregate gauges may be partial",
            snapshot.margin_failures
        );
    }

    debug!(
        "Risk metrics: {:.1}% margin util, {} near-liq, risk levels: safe={} warn={} danger={} crit={}",
        snapshot.margin_utilization(),
        snapshot.accounts_near_liquidation,
        snapshot.accounts_safe,
        snapshot.accounts_warning,
        snapshot.accounts_danger,
        snapshot.accounts_critical
    );
}

fn short_wallet(wallet: &WalletAddress) -> String {
    let wallet_str = wallet.to_string();
    if wallet_str.len() > 12 {
        format!(
            "{}...{}",
            &wallet_str[..6],
            &wallet_str[wallet_str.len() - 4..]
        )
    } else {
        wallet_str
    }
}