scirs2-integrate 0.4.3

//! Real-time risk monitoring and alerting
//!
//! This module provides real-time risk calculation, monitoring, and alerting capabilities
//! for trading systems and portfolio management applications.
//!
//! # Features
//! - Streaming risk calculations
//! - Real-time Greeks updates
//! - Position limit monitoring
//! - Risk limit breach detection
//! - Incremental VaR updates
//! - Low-latency risk aggregation

use crate::error::{IntegrateError, Result};
use crate::specialized::finance::pricing::black_scholes::normal_cdf;
use std::collections::{HashMap, VecDeque};

// ─────────────────────────────────────────────────────────────────────────────
// Core data types
// ─────────────────────────────────────────────────────────────────────────────

/// A single market data event (one price/volume update for one symbol).
#[derive(Debug, Clone)]
pub struct MarketTick {
    /// Ticker symbol (e.g., "AAPL", "SPX").
    pub symbol: String,
    /// Last-traded price.
    pub price: f64,
    /// Traded volume at this tick.
    pub volume: f64,
    /// Unix timestamp in milliseconds.
    pub timestamp: u64,
    /// Best bid price.
    pub bid: f64,
    /// Best ask price.
    pub ask: f64,
}

/// A risk-limit alert generated by one of the monitors.
#[derive(Debug, Clone)]
pub enum RiskAlert {
    /// A position-size limit has been exceeded.
    LimitBreach {
        /// Affected symbol.
        symbol: String,
        /// Current notional exposure (price × position).
        current: f64,
        /// Configured limit.
        limit: f64,
    },
    /// Portfolio VaR has exceeded its configured threshold.
    VarExceeded {
        /// Portfolio identifier.
        portfolio: String,
        /// Current estimated VaR.
        current_var: f64,
        /// VaR limit.
        limit: f64,
    },
    /// A Greek has breached its threshold.
    GreeksThreshold {
        /// Affected symbol / position label.
        symbol: String,
        /// Name of the Greek ("delta", "gamma", …).
        greek: String,
        /// Current absolute value of the Greek.
        value: f64,
    },
}

// ─────────────────────────────────────────────────────────────────────────────
// RiskMonitor trait
// ─────────────────────────────────────────────────────────────────────────────

/// A real-time risk monitor that reacts to market data ticks.
pub trait RiskMonitor: Send + Sync {
    /// Process one tick, optionally returning an alert.
    fn on_market_data(&mut self, tick: &MarketTick) -> Result<Option<RiskAlert>>;
    /// Human-readable name of this monitor.
    fn name(&self) -> &str;
}

// ─────────────────────────────────────────────────────────────────────────────
// StreamingGreeks
// ─────────────────────────────────────────────────────────────────────────────

/// Option position parameters needed for streaming Greeks.
#[derive(Debug, Clone)]
pub struct OptionPosition {
    /// Strike price.
    pub strike: f64,
    /// Time to expiry in years.
    pub time_to_expiry: f64,
    /// Implied volatility (annualised).
    pub volatility: f64,
    /// Risk-free rate.
    pub risk_free_rate: f64,
    /// Notional position size (signed: positive = long, negative = short).
    pub quantity: f64,
    /// Threshold for the absolute delta value above which an alert is fired.
    pub delta_alert_threshold: f64,
}

/// Standard-normal PDF φ(x) = exp(-x²/2) / √(2π).
fn normal_pdf(x: f64) -> f64 {
    (-0.5 * x * x).exp() / (2.0 * std::f64::consts::PI).sqrt()
}

/// Black-Scholes call Greeks computed inline from first principles.
///
/// Returns `(delta, gamma, vega, theta, rho)`.
///
/// Panics: never — guarded against degenerate inputs internally.
fn bs_call_greeks(
    spot: f64,
    strike: f64,
    time: f64,
    vol: f64,
    rate: f64,
) -> (f64, f64, f64, f64, f64) {
    if spot <= 0.0 || strike <= 0.0 || time <= 0.0 || vol <= 0.0 {
        return (0.0, 0.0, 0.0, 0.0, 0.0);
    }
    let sqrt_t = time.sqrt();
    let d1 = ((spot / strike).ln() + (rate + 0.5 * vol * vol) * time) / (vol * sqrt_t);
    let d2 = d1 - vol * sqrt_t;

    let nd1 = normal_cdf(d1);
    let nd2 = normal_cdf(d2);
    let phi_d1 = normal_pdf(d1);

    let delta = nd1;
    let gamma = phi_d1 / (spot * vol * sqrt_t);
    let vega = spot * phi_d1 * sqrt_t;
    let theta = -spot * phi_d1 * vol / (2.0 * sqrt_t) - rate * strike * (-rate * time).exp() * nd2;
    let rho = strike * time * (-rate * time).exp() * nd2;

    (delta, gamma, vega, theta, rho)
}

/// Maintains running Black-Scholes Greeks for a set of option positions,
/// updated on every market data tick for the relevant underlying.
#[derive(Debug)]
pub struct StreamingGreeks {
    /// Option positions keyed by underlying symbol.
    positions: HashMap<String, OptionPosition>,
    /// Most-recently computed Greeks, keyed by symbol.
    current_greeks: HashMap<String, (f64, f64, f64, f64, f64)>,
}

impl StreamingGreeks {
    /// Create a new instance with no positions.
    pub fn new() -> Self {
        Self {
            positions: HashMap::new(),
            current_greeks: HashMap::new(),
        }
    }

    /// Register an option position for a given underlying symbol.
    pub fn add_position(&mut self, symbol: String, pos: OptionPosition) {
        self.positions.insert(symbol, pos);
    }

    /// Return the latest Greeks for a symbol, or `None` if unknown.
    pub fn greeks_for(&self, symbol: &str) -> Option<(f64, f64, f64, f64, f64)> {
        self.current_greeks.get(symbol).copied()
    }
}

impl Default for StreamingGreeks {
    fn default() -> Self {
        Self::new()
    }
}

impl RiskMonitor for StreamingGreeks {
    fn name(&self) -> &str {
        "StreamingGreeks"
    }

    fn on_market_data(&mut self, tick: &MarketTick) -> Result<Option<RiskAlert>> {
        if let Some(pos) = self.positions.get(&tick.symbol) {
            let (delta, gamma, vega, theta, rho) = bs_call_greeks(
                tick.price,
                pos.strike,
                pos.time_to_expiry,
                pos.volatility,
                pos.risk_free_rate,
            );
            // Scale by position size
            let scaled_delta = delta * pos.quantity;
            let threshold = pos.delta_alert_threshold;
            self.current_greeks
                .insert(tick.symbol.clone(), (delta, gamma, vega, theta, rho));

            if scaled_delta.abs() > threshold {
                return Ok(Some(RiskAlert::GreeksThreshold {
                    symbol: tick.symbol.clone(),
                    greek: "delta".to_string(),
                    value: scaled_delta.abs(),
                }));
            }
        }
        Ok(None)
    }
}

// ─────────────────────────────────────────────────────────────────────────────
// PositionLimitChecker
// ─────────────────────────────────────────────────────────────────────────────

/// Checks whether the notional exposure (price × position) of any symbol
/// exceeds its configured limit, firing a `LimitBreach` alert on violations.
#[derive(Debug)]
pub struct PositionLimitChecker {
    /// symbol → maximum allowed notional exposure.
    limits: HashMap<String, f64>,
    /// symbol → current position size (in contracts / shares).
    current_positions: HashMap<String, f64>,
}

impl PositionLimitChecker {
    /// Create a new checker with empty limits and positions.
    pub fn new() -> Self {
        Self {
            limits: HashMap::new(),
            current_positions: HashMap::new(),
        }
    }

    /// Set the notional limit for a symbol.
    pub fn set_limit(&mut self, symbol: String, limit: f64) {
        self.limits.insert(symbol, limit);
    }

    /// Set the current position size for a symbol.
    pub fn set_position(&mut self, symbol: String, size: f64) {
        self.current_positions.insert(symbol, size);
    }
}

impl Default for PositionLimitChecker {
    fn default() -> Self {
        Self::new()
    }
}

impl RiskMonitor for PositionLimitChecker {
    fn name(&self) -> &str {
        "PositionLimitChecker"
    }

    fn on_market_data(&mut self, tick: &MarketTick) -> Result<Option<RiskAlert>> {
        let Some(&limit) = self.limits.get(&tick.symbol) else {
            return Ok(None);
        };
        let position = self
            .current_positions
            .get(&tick.symbol)
            .copied()
            .unwrap_or(0.0);
        let notional = tick.price * position.abs();

        if notional > limit {
            return Ok(Some(RiskAlert::LimitBreach {
                symbol: tick.symbol.clone(),
                current: notional,
                limit,
            }));
        }
        Ok(None)
    }
}

// ─────────────────────────────────────────────────────────────────────────────
// IncrementalVaR
// ─────────────────────────────────────────────────────────────────────────────

/// Maintains a rolling window of log-returns and computes the 95 % historical
/// VaR (5th-percentile loss, negated) after every tick.
///
/// A `VarExceeded` alert is emitted whenever the computed VaR breaches its
/// configured limit.
#[derive(Debug)]
pub struct IncrementalVaR {
    /// Sliding window of recent log-returns (oldest at front).
    window: VecDeque<f64>,
    /// Maximum window size (default: 252 trading days).
    window_size: usize,
    /// The VaR limit; alerts fire if VaR exceeds this.
    var_limit: f64,
    /// Portfolio name (used in alerts).
    portfolio: String,
    /// Previous prices, keyed by symbol (used to compute returns).
    prev_price: HashMap<String, f64>,
}

impl IncrementalVaR {
    /// Create with a given window size and VaR limit.
    pub fn new(portfolio: impl Into<String>, window_size: usize, var_limit: f64) -> Result<Self> {
        if window_size == 0 {
            return Err(IntegrateError::ValueError(
                "window_size must be positive".to_string(),
            ));
        }
        Ok(Self {
            window: VecDeque::with_capacity(window_size),
            window_size,
            var_limit,
            portfolio: portfolio.into(),
            prev_price: HashMap::new(),
        })
    }

    /// Compute the current 95 % historical VaR from the internal window.
    ///
    /// Returns `None` if the window has fewer than 2 observations.
    pub fn current_var(&self) -> Option<f64> {
        if self.window.len() < 2 {
            return None;
        }
        let mut sorted: Vec<f64> = self.window.iter().copied().collect();
        sorted.sort_by(|a, b| a.partial_cmp(b).unwrap_or(std::cmp::Ordering::Equal));
        // 5th percentile index (floor)
        let idx = ((1.0 - 0.95) * sorted.len() as f64).floor() as usize;
        let idx = idx.min(sorted.len().saturating_sub(1));
        Some(-sorted[idx])
    }
}

impl RiskMonitor for IncrementalVaR {
    fn name(&self) -> &str {
        "IncrementalVaR"
    }

    fn on_market_data(&mut self, tick: &MarketTick) -> Result<Option<RiskAlert>> {
        if tick.price <= 0.0 {
            return Err(IntegrateError::ValueError(format!(
                "Non-positive price {} for {}",
                tick.price, tick.symbol
            )));
        }
        // Compute log-return from the previous price for this symbol.
        if let Some(&prev) = self.prev_price.get(&tick.symbol) {
            if prev > 0.0 {
                let log_return = (tick.price / prev).ln();
                // Slide the window.
                if self.window.len() >= self.window_size {
                    self.window.pop_front();
                }
                self.window.push_back(log_return);
            }
        }
        self.prev_price.insert(tick.symbol.clone(), tick.price);

        // Check VaR limit.
        if let Some(var) = self.current_var() {
            if var > self.var_limit {
                return Ok(Some(RiskAlert::VarExceeded {
                    portfolio: self.portfolio.clone(),
                    current_var: var,
                    limit: self.var_limit,
                }));
            }
        }
        Ok(None)
    }
}

// ─────────────────────────────────────────────────────────────────────────────
// RiskAggregator
// ─────────────────────────────────────────────────────────────────────────────

/// Portfolio-level risk aggregation: sums notional exposures and VaR
/// contributions across all registered positions/symbols.
#[derive(Debug, Default)]
pub struct RiskAggregator {
    /// symbol → (position size, last price)
    positions: HashMap<String, (f64, f64)>,
    /// symbol → VaR contribution (updated externally or via `absorb_var`).
    var_contributions: HashMap<String, f64>,
}

impl RiskAggregator {
    /// Create an empty aggregator.
    pub fn new() -> Self {
        Self::default()
    }

    /// Register or update the position size and last price for a symbol.
    pub fn update_position(&mut self, symbol: String, size: f64, price: f64) {
        self.positions.insert(symbol, (size, price));
    }

    /// Set the VaR contribution for a symbol.
    pub fn set_var_contribution(&mut self, symbol: String, var: f64) {
        self.var_contributions.insert(symbol, var);
    }

    /// Total notional exposure (Σ |size| × price).
    pub fn total_notional(&self) -> f64 {
        self.positions
            .values()
            .map(|(size, price)| size.abs() * price)
            .sum()
    }

    /// Sum of VaR contributions (simple additive aggregation — conservative
    /// as it ignores cross-correlations).
    pub fn total_var(&self) -> f64 {
        self.var_contributions.values().copied().sum()
    }
}

impl RiskMonitor for RiskAggregator {
    fn name(&self) -> &str {
        "RiskAggregator"
    }

    fn on_market_data(&mut self, tick: &MarketTick) -> Result<Option<RiskAlert>> {
        // Update the last price for the symbol if we hold a position.
        if let Some(pos) = self.positions.get_mut(&tick.symbol) {
            pos.1 = tick.price;
        }
        Ok(None)
    }
}

// ─────────────────────────────────────────────────────────────────────────────
// EventDrivenRisk
// ─────────────────────────────────────────────────────────────────────────────

/// Dispatcher that fans out every `MarketTick` to all registered monitors and
/// collects the resulting alerts.
pub struct EventDrivenRisk {
    monitors: Vec<Box<dyn RiskMonitor>>,
}

impl EventDrivenRisk {
    /// Create an empty dispatcher.
    pub fn new() -> Self {
        Self {
            monitors: Vec::new(),
        }
    }

    /// Add a monitor to the dispatcher.
    pub fn add_monitor(&mut self, monitor: Box<dyn RiskMonitor>) {
        self.monitors.push(monitor);
    }

    /// Send one tick to all monitors and collect all alerts.
    ///
    /// Errors from individual monitors are surfaced as `IntegrateError`.
    pub fn process_tick(&mut self, tick: &MarketTick) -> Vec<RiskAlert> {
        let mut alerts = Vec::new();
        for monitor in &mut self.monitors {
            match monitor.on_market_data(tick) {
                Ok(Some(alert)) => alerts.push(alert),
                Ok(None) => {}
                Err(e) => {
                    // Non-fatal: log the monitor name and continue.
                    eprintln!("[EventDrivenRisk] monitor '{}' error: {e}", monitor.name());
                }
            }
        }
        alerts
    }
}

impl Default for EventDrivenRisk {
    fn default() -> Self {
        Self::new()
    }
}

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

#[cfg(test)]
mod tests {
    use super::*;

    fn make_tick(symbol: &str, price: f64) -> MarketTick {
        MarketTick {
            symbol: symbol.to_string(),
            price,
            volume: 1000.0,
            timestamp: 1_700_000_000_000,
            bid: price - 0.01,
            ask: price + 0.01,
        }
    }

    // ──────────────────────────────────────────────────────────────────────
    // Test 1: PositionLimitChecker fires LimitBreach when limit exceeded
    // ──────────────────────────────────────────────────────────────────────
    #[test]
    fn test_position_limit_breach() {
        let mut checker = PositionLimitChecker::new();
        // Allow up to $5 000 notional exposure for AAPL.
        checker.set_limit("AAPL".to_string(), 5_000.0);
        // Hold 100 shares.
        checker.set_position("AAPL".to_string(), 100.0);

        // Price = $60 → notional = $6 000 > $5 000 → breach.
        let tick = make_tick("AAPL", 60.0);
        let alert = checker
            .on_market_data(&tick)
            .expect("monitor should succeed");

        match alert {
            Some(RiskAlert::LimitBreach {
                symbol,
                current,
                limit,
            }) => {
                assert_eq!(symbol, "AAPL");
                assert!((current - 6_000.0).abs() < 1e-9, "current={current}");
                assert!((limit - 5_000.0).abs() < 1e-9, "limit={limit}");
            }
            other => panic!("Expected LimitBreach, got {other:?}"),
        }
    }

    /// No alert when notional is within limits.
    #[test]
    fn test_position_limit_no_breach() {
        let mut checker = PositionLimitChecker::new();
        checker.set_limit("AAPL".to_string(), 5_000.0);
        checker.set_position("AAPL".to_string(), 100.0);

        // Price = $40 → notional = $4 000 < $5 000 → no alert.
        let tick = make_tick("AAPL", 40.0);
        let alert = checker
            .on_market_data(&tick)
            .expect("monitor should succeed");

        assert!(alert.is_none(), "Expected no alert, got {alert:?}");
    }

    // ──────────────────────────────────────────────────────────────────────
    // Test 2: IncrementalVaR returns correct VaR for known return sequence
    // ──────────────────────────────────────────────────────────────────────
    #[test]
    fn test_incremental_var_known_sequence() {
        // Feed 20 prices that imply a 5-percent loss at the 5th percentile.
        // We use an arithmetic sequence: p_0 = 100, then alternate +1/-5.
        // After sufficient ticks the window should contain losses of ~5 %.
        let mut ivar = IncrementalVaR::new("TEST", 252, 0.0).expect("should construct");

        // Synthetic prices: start at 100, then drop by 5 for one step and
        // recover, so we get a known worst-case return.
        let prices: Vec<f64> = vec![
            100.0, 95.0, 100.0, 95.0, 100.0, 95.0, 100.0, 95.0, 100.0, 95.0, 100.0, 95.0, 100.0,
            95.0, 100.0, 95.0, 100.0, 95.0, 100.0, 95.0,
        ];

        for &p in &prices {
            let tick = make_tick("SPX", p);
            let _ = ivar.on_market_data(&tick).expect("tick should succeed");
        }

        let var = ivar.current_var().expect("should have enough data");
        // The negative log-returns here are ln(100/95)≈0.051 and
        // the positive ones are ln(95/100)≈-0.051. The 5th percentile of a
        // window that alternates equally between -0.051 and +0.051 is the
        // minimum, so VaR ≈ 0.051.
        assert!(var > 0.0, "VaR must be positive, got {var}");
        assert!(var < 0.2, "VaR seems too large: {var}");
    }

    // ──────────────────────────────────────────────────────────────────────
    // Test 3: EventDrivenRisk dispatches to multiple monitors
    // ──────────────────────────────────────────────────────────────────────
    #[test]
    fn test_event_driven_dispatches_to_multiple_monitors() {
        let mut checker1 = PositionLimitChecker::new();
        checker1.set_limit("X".to_string(), 1.0); // very small limit → always breach
        checker1.set_position("X".to_string(), 10.0);

        let mut checker2 = PositionLimitChecker::new();
        checker2.set_limit("X".to_string(), 1.0);
        checker2.set_position("X".to_string(), 10.0);

        let mut dispatcher = EventDrivenRisk::new();
        dispatcher.add_monitor(Box::new(checker1));
        dispatcher.add_monitor(Box::new(checker2));

        let tick = make_tick("X", 100.0);
        let alerts = dispatcher.process_tick(&tick);

        assert_eq!(
            alerts.len(),
            2,
            "Both monitors should have fired an alert, got {}",
            alerts.len()
        );
    }

    // ──────────────────────────────────────────────────────────────────────
    // Test 4: StreamingGreeks delta is in (0, 1) for a call option
    // ──────────────────────────────────────────────────────────────────────
    #[test]
    fn test_streaming_greeks_delta_plausible() {
        let mut sg = StreamingGreeks::new();
        sg.add_position(
            "AAPL".to_string(),
            OptionPosition {
                strike: 150.0,
                time_to_expiry: 0.25, // 3 months
                volatility: 0.20,
                risk_free_rate: 0.05,
                quantity: 1.0,
                delta_alert_threshold: f64::INFINITY, // no alert for this test
            },
        );

        let tick = make_tick("AAPL", 155.0);
        let _ = sg.on_market_data(&tick).expect("tick should succeed");

        let (delta, gamma, vega, _theta, _rho) = sg
            .greeks_for("AAPL")
            .expect("Greeks should be available after tick");

        // Call delta must be strictly between 0 and 1.
        assert!(delta > 0.0 && delta < 1.0, "delta out of range: {delta}");
        // Gamma is always non-negative for vanilla options.
        assert!(gamma >= 0.0, "gamma negative: {gamma}");
        // Vega is always non-negative.
        assert!(vega >= 0.0, "vega negative: {vega}");
    }

    // ──────────────────────────────────────────────────────────────────────
    // Test 5: Full pipeline — create all monitors, feed 20 ticks, collect alerts
    // ──────────────────────────────────────────────────────────────────────
    #[test]
    fn test_full_pipeline() {
        let mut dispatcher = EventDrivenRisk::new();

        // Position limit checker: limit = 1 000, position = 20 shares.
        // At price > 50 → notional > 1 000 → breach.
        let mut checker = PositionLimitChecker::new();
        checker.set_limit("SPY".to_string(), 1_000.0);
        checker.set_position("SPY".to_string(), 20.0);
        dispatcher.add_monitor(Box::new(checker));

        // Incremental VaR (very low limit → triggers when loss exists).
        let ivar = IncrementalVaR::new("portfolio", 252, -f64::INFINITY).expect("should construct");
        dispatcher.add_monitor(Box::new(ivar));

        // Streaming Greeks.
        let mut sg = StreamingGreeks::new();
        sg.add_position(
            "SPY".to_string(),
            OptionPosition {
                strike: 400.0,
                time_to_expiry: 0.5,
                volatility: 0.18,
                risk_free_rate: 0.04,
                quantity: 1.0,
                delta_alert_threshold: f64::INFINITY,
            },
        );
        dispatcher.add_monitor(Box::new(sg));

        // Risk aggregator.
        let mut agg = RiskAggregator::new();
        agg.update_position("SPY".to_string(), 20.0, 420.0);
        dispatcher.add_monitor(Box::new(agg));

        // Feed 20 ticks at price 60 (notional = 1 200 → above limit of 1 000).
        let mut total_alerts = 0usize;
        for i in 0..20u64 {
            let tick = MarketTick {
                symbol: "SPY".to_string(),
                price: 60.0,
                volume: 500.0,
                timestamp: 1_700_000_000_000 + i * 1_000,
                bid: 59.99,
                ask: 60.01,
            };
            let alerts = dispatcher.process_tick(&tick);
            total_alerts += alerts.len();
        }

        // At a minimum the position limit checker should have fired 20 times.
        assert!(
            total_alerts >= 20,
            "Expected at least 20 alerts across 20 ticks, got {total_alerts}"
        );
    }
}