e4870b17b9
Reduce log noise for non-critical operational paths: connection retries, expected fallbacks, graceful degradation, and optional feature absence. Keeps warn/error for genuine failures requiring attention. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
1079 lines
36 KiB
Rust
1079 lines
36 KiB
Rust
//! Position Limits Repository
|
|
//!
|
|
//! Manages position limits tracking, enforcement, and breach detection for
|
|
//! high-frequency trading systems with real-time monitoring capabilities.
|
|
|
|
// Allow pedantic lints for limits implementation
|
|
#![allow(clippy::missing_docs_in_private_items)]
|
|
#![allow(missing_docs)]
|
|
|
|
use async_trait::async_trait;
|
|
use chrono::{DateTime, Utc};
|
|
use redis::aio::ConnectionManager;
|
|
use rust_decimal::Decimal;
|
|
use serde::{Deserialize, Serialize};
|
|
use sqlx::{PgPool, Row};
|
|
use std::collections::HashMap;
|
|
use tracing::{error, info, warn};
|
|
use uuid::Uuid;
|
|
|
|
use crate::{RiskDataError, RiskDataResult};
|
|
|
|
/// Position limit types
|
|
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, sqlx::Type)]
|
|
#[sqlx(type_name = "limit_type", rename_all = "snake_case")]
|
|
pub enum LimitType {
|
|
/// Maximum position size limit
|
|
MaxPosition,
|
|
/// Maximum daily trading volume limit
|
|
MaxDailyVolume,
|
|
/// Maximum daily loss limit
|
|
MaxDailyLoss,
|
|
/// Maximum drawdown limit
|
|
MaxDrawdown,
|
|
/// Value-at-Risk based limit
|
|
VarLimit,
|
|
/// Portfolio concentration limit
|
|
ConcentrationLimit,
|
|
/// Maximum leverage limit
|
|
LeverageLimit,
|
|
/// Market exposure limit
|
|
ExposureLimit,
|
|
}
|
|
|
|
/// Limit scope (what the limit applies to)
|
|
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, sqlx::Type)]
|
|
#[sqlx(type_name = "limit_scope", rename_all = "snake_case")]
|
|
pub enum LimitScope {
|
|
/// System-wide limit applying to all positions
|
|
Global,
|
|
/// Portfolio-specific limit
|
|
Portfolio,
|
|
/// Strategy-specific limit
|
|
Strategy,
|
|
/// Symbol-specific limit
|
|
Symbol,
|
|
/// Sector-specific limit
|
|
Sector,
|
|
/// Asset class specific limit
|
|
AssetClass,
|
|
/// Trader-specific limit
|
|
Trader,
|
|
}
|
|
|
|
/// Limit breach severity
|
|
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, sqlx::Type)]
|
|
#[sqlx(type_name = "breach_severity", rename_all = "snake_case")]
|
|
pub enum BreachSeverity {
|
|
/// Approaching limit (80-90%)
|
|
Warning,
|
|
/// Soft breach (90-100%)
|
|
Soft,
|
|
/// Hard breach (>100%)
|
|
Hard,
|
|
/// Critical breach (>120%)
|
|
Critical,
|
|
}
|
|
|
|
/// Position limit definition
|
|
#[derive(Debug, Clone, Serialize, Deserialize, sqlx::FromRow)]
|
|
pub struct PositionLimit {
|
|
/// Unique identifier for this position limit
|
|
pub id: Uuid,
|
|
/// Human-readable name for this limit
|
|
pub name: String,
|
|
/// Type of limit being enforced
|
|
pub limit_type: LimitType,
|
|
/// Scope that this limit applies to
|
|
pub scope: LimitScope,
|
|
/// Specific value for the scope (Portfolio ID, symbol, etc.)
|
|
pub scope_value: Option<String>,
|
|
/// Maximum threshold value for this limit
|
|
pub threshold: Decimal,
|
|
/// Optional warning level threshold
|
|
pub warning_threshold: Option<Decimal>,
|
|
/// Currency of the limit (ISO 3-letter code)
|
|
pub currency: String,
|
|
/// Whether this limit is currently enabled
|
|
pub enabled: bool,
|
|
/// Timestamp when the limit was created
|
|
pub created_at: DateTime<Utc>,
|
|
/// Timestamp when the limit was last updated
|
|
pub updated_at: DateTime<Utc>,
|
|
/// User who created this limit
|
|
pub created_by: String,
|
|
/// Optional description of the limit
|
|
pub description: Option<String>,
|
|
/// Additional limit metadata as JSON
|
|
pub metadata: serde_json::Value,
|
|
}
|
|
|
|
/// Current position exposure
|
|
#[derive(Debug, Clone, Serialize, Deserialize, sqlx::FromRow)]
|
|
pub struct PositionExposure {
|
|
/// Scope of this exposure measurement
|
|
pub scope: LimitScope,
|
|
/// Specific value for the scope
|
|
pub scope_value: String,
|
|
/// Symbol for symbol-specific exposures
|
|
pub symbol: Option<String>,
|
|
/// Current position size (shares/contracts)
|
|
pub current_position: Decimal,
|
|
/// Current market value of the position
|
|
pub current_value: Decimal,
|
|
/// Total daily trading volume
|
|
pub daily_volume: Decimal,
|
|
/// Daily profit and loss
|
|
pub daily_pnl: Decimal,
|
|
/// Currency of the exposure (ISO 3-letter code)
|
|
pub currency: String,
|
|
/// Timestamp when this exposure was last updated
|
|
pub updated_at: DateTime<Utc>,
|
|
}
|
|
|
|
/// Limit breach event
|
|
#[derive(Debug, Clone, Serialize, Deserialize, sqlx::FromRow)]
|
|
pub struct LimitBreach {
|
|
/// Unique identifier for this breach event
|
|
pub id: Uuid,
|
|
/// Limit that was breached
|
|
pub limit_id: Uuid,
|
|
/// Severity level of the breach
|
|
pub severity: BreachSeverity,
|
|
/// Timestamp when the breach occurred
|
|
pub breach_time: DateTime<Utc>,
|
|
/// Current value that caused the breach
|
|
pub current_value: Decimal,
|
|
/// Limit threshold that was exceeded
|
|
pub limit_threshold: Decimal,
|
|
/// Percentage by which the limit was breached
|
|
pub breach_percentage: Decimal,
|
|
/// Scope where the breach occurred
|
|
pub scope: LimitScope,
|
|
/// Specific scope value for the breach
|
|
pub scope_value: Option<String>,
|
|
/// Symbol involved in the breach (if applicable)
|
|
pub symbol: Option<String>,
|
|
/// Portfolio ID associated with the breach
|
|
pub portfolio_id: Option<String>,
|
|
/// Strategy ID associated with the breach
|
|
pub strategy_id: Option<String>,
|
|
/// Trader ID associated with the breach
|
|
pub trader_id: Option<String>,
|
|
/// Whether the breach has been acknowledged
|
|
pub acknowledged: bool,
|
|
/// User who acknowledged the breach
|
|
pub acknowledged_by: Option<String>,
|
|
/// Timestamp when the breach was acknowledged
|
|
pub acknowledged_at: Option<DateTime<Utc>>,
|
|
/// Whether the breach has been resolved
|
|
pub resolved: bool,
|
|
/// Timestamp when the breach was resolved
|
|
pub resolved_at: Option<DateTime<Utc>>,
|
|
/// Note explaining the resolution
|
|
pub resolution_note: Option<String>,
|
|
/// Action taken to resolve the breach
|
|
pub action_taken: Option<String>,
|
|
}
|
|
|
|
/// Limit check request
|
|
#[derive(Debug, Clone, Serialize, Deserialize)]
|
|
pub struct LimitCheckRequest {
|
|
/// Scope to check limits for
|
|
pub scope: LimitScope,
|
|
/// Specific scope value
|
|
pub scope_value: String,
|
|
/// Symbol for the proposed trade (if applicable)
|
|
pub symbol: Option<String>,
|
|
/// Proposed position size change
|
|
pub proposed_position: Decimal,
|
|
/// Proposed value change
|
|
pub proposed_value: Decimal,
|
|
/// Currency of the proposed transaction (ISO 3-letter code)
|
|
pub currency: String,
|
|
}
|
|
|
|
/// Limit check result
|
|
#[derive(Debug, Clone, Serialize, Deserialize)]
|
|
pub struct LimitCheckResult {
|
|
/// Whether the proposed transaction is allowed
|
|
pub allowed: bool,
|
|
/// List of limit violations that would occur
|
|
pub violations: Vec<LimitViolation>,
|
|
/// List of warning-level limit breaches
|
|
pub warnings: Vec<LimitViolation>,
|
|
/// Maximum allowed position size (if restricted)
|
|
pub max_allowed_position: Option<Decimal>,
|
|
/// Maximum allowed value (if restricted)
|
|
pub max_allowed_value: Option<Decimal>,
|
|
}
|
|
|
|
/// Individual limit violation
|
|
#[derive(Debug, Clone, Serialize, Deserialize)]
|
|
pub struct LimitViolation {
|
|
/// ID of the limit that was violated
|
|
pub limit_id: Uuid,
|
|
/// Name of the limit that was violated
|
|
pub limit_name: String,
|
|
/// Type of the limit that was violated
|
|
pub limit_type: LimitType,
|
|
/// Current value that violates the limit
|
|
pub current_value: Decimal,
|
|
/// Threshold of the violated limit
|
|
pub limit_threshold: Decimal,
|
|
/// Percentage by which the limit was breached
|
|
pub breach_percentage: Decimal,
|
|
/// Severity of the violation
|
|
pub severity: BreachSeverity,
|
|
}
|
|
|
|
/// Limits repository trait
|
|
#[async_trait]
|
|
#[allow(clippy::module_name_repetitions)] // Repository pattern naming is conventional
|
|
pub trait LimitsRepository: Send + Sync + std::fmt::Debug {
|
|
/// Create or update a position limit
|
|
async fn upsert_limit(&self, limit: PositionLimit) -> RiskDataResult<()>;
|
|
|
|
/// Get all limits for a scope
|
|
async fn get_limits(
|
|
&self,
|
|
scope: LimitScope,
|
|
scope_value: Option<String>,
|
|
) -> RiskDataResult<Vec<PositionLimit>>;
|
|
|
|
/// Get a specific limit by ID
|
|
async fn get_limit(&self, limit_id: Uuid) -> RiskDataResult<Option<PositionLimit>>;
|
|
|
|
/// Delete a limit
|
|
async fn delete_limit(&self, limit_id: Uuid) -> RiskDataResult<()>;
|
|
|
|
/// Update position exposure
|
|
async fn update_exposure(&self, exposure: PositionExposure) -> RiskDataResult<()>;
|
|
|
|
/// Get current exposure for a scope
|
|
async fn get_exposure(
|
|
&self,
|
|
scope: LimitScope,
|
|
scope_value: &str,
|
|
) -> RiskDataResult<Option<PositionExposure>>;
|
|
|
|
/// Check if a proposed position would violate limits
|
|
async fn check_limits(&self, request: LimitCheckRequest) -> RiskDataResult<LimitCheckResult>;
|
|
|
|
/// Record a limit breach
|
|
async fn record_breach(&self, breach: LimitBreach) -> RiskDataResult<()>;
|
|
|
|
/// Get recent breaches
|
|
async fn get_breaches(
|
|
&self,
|
|
from: DateTime<Utc>,
|
|
to: DateTime<Utc>,
|
|
severity: Option<BreachSeverity>,
|
|
) -> RiskDataResult<Vec<LimitBreach>>;
|
|
|
|
/// Acknowledge a breach
|
|
async fn acknowledge_breach(
|
|
&self,
|
|
breach_id: Uuid,
|
|
acknowledged_by: String,
|
|
) -> RiskDataResult<()>;
|
|
|
|
/// Resolve a breach
|
|
async fn resolve_breach(
|
|
&self,
|
|
breach_id: Uuid,
|
|
resolution_note: String,
|
|
action_taken: String,
|
|
) -> RiskDataResult<()>;
|
|
|
|
/// Get limit utilization summary
|
|
async fn get_utilization_summary(&self) -> RiskDataResult<HashMap<String, Decimal>>;
|
|
|
|
/// Perform real-time limit monitoring
|
|
async fn monitor_limits(&self) -> RiskDataResult<Vec<LimitBreach>>;
|
|
}
|
|
|
|
/// Limits repository implementation
|
|
#[derive(Clone)]
|
|
#[allow(clippy::module_name_repetitions)] // Repository pattern naming is conventional
|
|
pub struct LimitsRepositoryImpl {
|
|
db_pool: PgPool,
|
|
redis_conn: ConnectionManager,
|
|
}
|
|
|
|
impl std::fmt::Debug for LimitsRepositoryImpl {
|
|
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
|
f.debug_struct("LimitsRepositoryImpl")
|
|
.field("db_pool", &"<PgPool>")
|
|
.field("redis_conn", &"<ConnectionManager>")
|
|
.finish()
|
|
}
|
|
}
|
|
|
|
impl LimitsRepositoryImpl {
|
|
pub const fn new(db_pool: PgPool, redis_conn: ConnectionManager) -> Self {
|
|
Self {
|
|
db_pool,
|
|
redis_conn,
|
|
}
|
|
}
|
|
|
|
/// Validate limit configuration
|
|
fn validate_limit(limit: &PositionLimit) -> RiskDataResult<()> {
|
|
if limit.name.is_empty() {
|
|
return Err(RiskDataError::LimitsValidation(
|
|
"Limit name cannot be empty".to_owned(),
|
|
));
|
|
}
|
|
|
|
if limit.threshold <= Decimal::ZERO {
|
|
return Err(RiskDataError::LimitsValidation(
|
|
"Limit threshold must be positive".to_owned(),
|
|
));
|
|
}
|
|
|
|
// Validate warning threshold if provided
|
|
if let Some(warning) = limit.warning_threshold {
|
|
if warning <= Decimal::ZERO || warning >= limit.threshold {
|
|
return Err(RiskDataError::LimitsValidation(
|
|
"Warning threshold must be positive and less than limit threshold".to_owned(),
|
|
));
|
|
}
|
|
}
|
|
|
|
// Validate scope-specific requirements
|
|
match limit.scope {
|
|
LimitScope::Portfolio
|
|
| LimitScope::Strategy
|
|
| LimitScope::Symbol
|
|
| LimitScope::Sector
|
|
| LimitScope::AssetClass
|
|
| LimitScope::Trader => match &limit.scope_value {
|
|
None => {
|
|
return Err(RiskDataError::LimitsValidation(format!(
|
|
"Scope value required for {:?} limits",
|
|
limit.scope
|
|
)));
|
|
},
|
|
Some(value) if value.is_empty() => {
|
|
return Err(RiskDataError::LimitsValidation(format!(
|
|
"Scope value required for {:?} limits",
|
|
limit.scope
|
|
)));
|
|
},
|
|
Some(_) => {},
|
|
},
|
|
LimitScope::Global => {
|
|
// Global limits don't require scope_value
|
|
},
|
|
}
|
|
|
|
Ok(())
|
|
}
|
|
|
|
/// Calculate breach severity based on percentage
|
|
fn calculate_breach_severity(breach_percentage: Decimal) -> BreachSeverity {
|
|
if breach_percentage >= Decimal::from(120) {
|
|
BreachSeverity::Critical
|
|
} else if breach_percentage >= Decimal::from(100) {
|
|
BreachSeverity::Hard
|
|
} else if breach_percentage >= Decimal::from(90) {
|
|
BreachSeverity::Soft
|
|
} else {
|
|
BreachSeverity::Warning
|
|
}
|
|
}
|
|
|
|
/// Check a single limit against current exposure
|
|
#[allow(clippy::arithmetic_side_effects)] // Financial calculations with Decimal are safe
|
|
#[allow(clippy::default_numeric_fallback)] // Type inference for numeric literals is acceptable here
|
|
async fn check_single_limit(
|
|
&self,
|
|
limit: &PositionLimit,
|
|
current_exposure: &PositionExposure,
|
|
proposed_value: Decimal,
|
|
) -> RiskDataResult<Option<LimitViolation>> {
|
|
if !limit.enabled {
|
|
return Ok(None);
|
|
}
|
|
|
|
let test_value = match limit.limit_type {
|
|
LimitType::MaxPosition => current_exposure.current_position + proposed_value,
|
|
LimitType::MaxDailyVolume => current_exposure.daily_volume + proposed_value.abs(),
|
|
LimitType::MaxDailyLoss => {
|
|
if current_exposure.daily_pnl < Decimal::ZERO {
|
|
current_exposure.daily_pnl.abs() // Current loss
|
|
} else {
|
|
Decimal::ZERO
|
|
}
|
|
},
|
|
LimitType::ExposureLimit => current_exposure.current_value + proposed_value,
|
|
LimitType::MaxDrawdown
|
|
| LimitType::VarLimit
|
|
| LimitType::ConcentrationLimit
|
|
| LimitType::LeverageLimit => {
|
|
tracing::warn!("Unknown limit type in exposure calculation - using current value as conservative fallback");
|
|
current_exposure.current_value // Conservative fallback for unknown limit types
|
|
},
|
|
};
|
|
|
|
let breach_percentage = (test_value / limit.threshold) * Decimal::from(100);
|
|
|
|
// Check for breach or warning
|
|
let threshold_to_check = if let Some(warning) = limit.warning_threshold {
|
|
if breach_percentage >= (warning / limit.threshold) * Decimal::from(100) {
|
|
warning
|
|
} else {
|
|
return Ok(None); // No violation
|
|
}
|
|
} else {
|
|
limit.threshold
|
|
};
|
|
|
|
if test_value > threshold_to_check {
|
|
let severity = if test_value > limit.threshold {
|
|
Self::calculate_breach_severity(breach_percentage)
|
|
} else {
|
|
BreachSeverity::Warning
|
|
};
|
|
|
|
Ok(Some(LimitViolation {
|
|
limit_id: limit.id,
|
|
limit_name: limit.name.clone(),
|
|
limit_type: limit.limit_type,
|
|
current_value: test_value,
|
|
limit_threshold: limit.threshold,
|
|
breach_percentage,
|
|
severity,
|
|
}))
|
|
} else {
|
|
Ok(None)
|
|
}
|
|
}
|
|
}
|
|
|
|
#[async_trait]
|
|
impl LimitsRepository for LimitsRepositoryImpl {
|
|
#[allow(clippy::as_conversions)] // Safe enum discriminant to u8 for cache keys
|
|
async fn upsert_limit(&self, limit: PositionLimit) -> RiskDataResult<()> {
|
|
Self::validate_limit(&limit)?;
|
|
|
|
let query = "
|
|
INSERT INTO position_limits (
|
|
id, name, limit_type, scope, scope_value, threshold, warning_threshold,
|
|
currency, enabled, created_at, updated_at, created_by, description, metadata
|
|
) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14)
|
|
ON CONFLICT (id) DO UPDATE SET
|
|
name = EXCLUDED.name,
|
|
threshold = EXCLUDED.threshold,
|
|
warning_threshold = EXCLUDED.warning_threshold,
|
|
enabled = EXCLUDED.enabled,
|
|
updated_at = EXCLUDED.updated_at,
|
|
description = EXCLUDED.description,
|
|
metadata = EXCLUDED.metadata
|
|
";
|
|
|
|
sqlx::query(query)
|
|
.bind(limit.id)
|
|
.bind(&limit.name)
|
|
.bind(limit.limit_type)
|
|
.bind(limit.scope)
|
|
.bind(&limit.scope_value)
|
|
.bind(limit.threshold)
|
|
.bind(limit.warning_threshold)
|
|
.bind(&limit.currency)
|
|
.bind(limit.enabled)
|
|
.bind(limit.created_at)
|
|
.bind(limit.updated_at)
|
|
.bind(&limit.created_by)
|
|
.bind(&limit.description)
|
|
.bind(&limit.metadata)
|
|
.execute(&self.db_pool)
|
|
.await?;
|
|
|
|
// Cache active limits in Redis for fast access
|
|
if limit.enabled {
|
|
let cache_key = format!(
|
|
"limits:{}:{}",
|
|
limit.scope as u8,
|
|
limit.scope_value.as_deref().unwrap_or("global")
|
|
);
|
|
let serialized = serde_json::to_string(&limit)?;
|
|
|
|
let mut redis_conn = self.redis_conn.clone();
|
|
redis::cmd("HSET")
|
|
.arg(&cache_key)
|
|
.arg(limit.id.to_string())
|
|
.arg(&serialized)
|
|
.query_async::<()>(&mut redis_conn)
|
|
.await?;
|
|
}
|
|
|
|
info!(
|
|
"Position limit upserted: {} - {}",
|
|
limit.name, limit.threshold
|
|
);
|
|
Ok(())
|
|
}
|
|
|
|
async fn get_limits(
|
|
&self,
|
|
scope: LimitScope,
|
|
scope_value: Option<String>,
|
|
) -> RiskDataResult<Vec<PositionLimit>> {
|
|
let query = if scope_value.is_some() {
|
|
"SELECT * FROM position_limits WHERE scope = $1 AND scope_value = $2 ORDER BY name"
|
|
} else {
|
|
"SELECT * FROM position_limits WHERE scope = $1 AND scope_value IS NULL ORDER BY name"
|
|
};
|
|
|
|
let limits = if let Some(sv) = scope_value {
|
|
sqlx::query_as::<_, PositionLimit>(query)
|
|
.bind(scope)
|
|
.bind(sv)
|
|
.fetch_all(&self.db_pool)
|
|
.await?
|
|
} else {
|
|
sqlx::query_as::<_, PositionLimit>(query)
|
|
.bind(scope)
|
|
.fetch_all(&self.db_pool)
|
|
.await?
|
|
};
|
|
|
|
Ok(limits)
|
|
}
|
|
|
|
async fn get_limit(&self, limit_id: Uuid) -> RiskDataResult<Option<PositionLimit>> {
|
|
let query = "SELECT * FROM position_limits WHERE id = $1";
|
|
|
|
let limit = sqlx::query_as::<_, PositionLimit>(query)
|
|
.bind(limit_id)
|
|
.fetch_optional(&self.db_pool)
|
|
.await?;
|
|
|
|
Ok(limit)
|
|
}
|
|
|
|
async fn delete_limit(&self, limit_id: Uuid) -> RiskDataResult<()> {
|
|
// Remove from database
|
|
let query = "DELETE FROM position_limits WHERE id = $1";
|
|
sqlx::query(query)
|
|
.bind(limit_id)
|
|
.execute(&self.db_pool)
|
|
.await?;
|
|
|
|
// Remove from Redis cache
|
|
let mut redis_conn = self.redis_conn.clone();
|
|
let cache_keys: Vec<String> = redis::cmd("KEYS")
|
|
.arg("limits:*")
|
|
.query_async(&mut redis_conn)
|
|
.await?;
|
|
|
|
for key in cache_keys {
|
|
redis::cmd("HDEL")
|
|
.arg(&key)
|
|
.arg(limit_id.to_string())
|
|
.query_async::<()>(&mut redis_conn)
|
|
.await?;
|
|
}
|
|
|
|
info!("Position limit deleted: {}", limit_id);
|
|
Ok(())
|
|
}
|
|
|
|
#[allow(clippy::as_conversions)] // Safe enum discriminant to u8 for cache keys
|
|
#[allow(clippy::default_numeric_fallback)] // TTL duration literal
|
|
async fn update_exposure(&self, exposure: PositionExposure) -> RiskDataResult<()> {
|
|
let query = "
|
|
INSERT INTO position_exposures (
|
|
scope, scope_value, symbol, current_position, current_value,
|
|
daily_volume, daily_pnl, currency, updated_at
|
|
) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
|
|
ON CONFLICT (scope, scope_value, COALESCE(symbol, '')) DO UPDATE SET
|
|
current_position = EXCLUDED.current_position,
|
|
current_value = EXCLUDED.current_value,
|
|
daily_volume = EXCLUDED.daily_volume,
|
|
daily_pnl = EXCLUDED.daily_pnl,
|
|
updated_at = EXCLUDED.updated_at
|
|
";
|
|
|
|
sqlx::query(query)
|
|
.bind(exposure.scope)
|
|
.bind(&exposure.scope_value)
|
|
.bind(&exposure.symbol)
|
|
.bind(exposure.current_position)
|
|
.bind(exposure.current_value)
|
|
.bind(exposure.daily_volume)
|
|
.bind(exposure.daily_pnl)
|
|
.bind(&exposure.currency)
|
|
.bind(exposure.updated_at)
|
|
.execute(&self.db_pool)
|
|
.await?;
|
|
|
|
// Cache current exposure in Redis for real-time access
|
|
let cache_key = format!("exposure:{}:{}", exposure.scope as u8, exposure.scope_value);
|
|
let serialized = serde_json::to_string(&exposure)?;
|
|
|
|
let mut redis_conn = self.redis_conn.clone();
|
|
redis::cmd("SETEX")
|
|
.arg(&cache_key)
|
|
.arg(300) // 5 minutes TTL
|
|
.arg(&serialized)
|
|
.query_async::<()>(&mut redis_conn)
|
|
.await?;
|
|
|
|
Ok(())
|
|
}
|
|
|
|
#[allow(clippy::as_conversions)] // Safe enum discriminant to u8 for cache keys
|
|
async fn get_exposure(
|
|
&self,
|
|
scope: LimitScope,
|
|
scope_value: &str,
|
|
) -> RiskDataResult<Option<PositionExposure>> {
|
|
// Try Redis cache first
|
|
let cache_key = format!("exposure:{}:{}", scope as u8, scope_value);
|
|
let mut redis_conn = self.redis_conn.clone();
|
|
|
|
if let Ok(cached) = redis::cmd("GET")
|
|
.arg(&cache_key)
|
|
.query_async::<String>(&mut redis_conn)
|
|
.await
|
|
{
|
|
if let Ok(exposure) = serde_json::from_str::<PositionExposure>(&cached) {
|
|
return Ok(Some(exposure));
|
|
}
|
|
}
|
|
|
|
// Fallback to database
|
|
let query = "SELECT * FROM position_exposures WHERE scope = $1 AND scope_value = $2";
|
|
|
|
let exposure = sqlx::query_as::<_, PositionExposure>(query)
|
|
.bind(scope)
|
|
.bind(scope_value)
|
|
.fetch_optional(&self.db_pool)
|
|
.await?;
|
|
|
|
Ok(exposure)
|
|
}
|
|
|
|
#[allow(clippy::arithmetic_side_effects)] // Financial calculations with Decimal are safe
|
|
#[allow(clippy::default_numeric_fallback)] // Type inference for numeric literals is acceptable here
|
|
async fn check_limits(&self, request: LimitCheckRequest) -> RiskDataResult<LimitCheckResult> {
|
|
// Get applicable limits
|
|
let mut all_limits = Vec::new();
|
|
|
|
// Global limits
|
|
all_limits.extend(self.get_limits(LimitScope::Global, None).await?);
|
|
|
|
// Scope-specific limits
|
|
all_limits.extend(
|
|
self.get_limits(request.scope, Some(request.scope_value.clone()))
|
|
.await?,
|
|
);
|
|
|
|
// Symbol-specific limits if symbol provided
|
|
if let Some(symbol) = &request.symbol {
|
|
all_limits.extend(
|
|
self.get_limits(LimitScope::Symbol, Some(symbol.clone()))
|
|
.await?,
|
|
);
|
|
}
|
|
|
|
// Get current exposure
|
|
let current_exposure = self
|
|
.get_exposure(request.scope, &request.scope_value)
|
|
.await?
|
|
.unwrap_or_else(|| PositionExposure {
|
|
scope: request.scope,
|
|
scope_value: request.scope_value.clone(),
|
|
symbol: request.symbol.clone(),
|
|
current_position: Decimal::ZERO,
|
|
current_value: Decimal::ZERO,
|
|
daily_volume: Decimal::ZERO,
|
|
daily_pnl: Decimal::ZERO,
|
|
currency: request.currency.clone(),
|
|
updated_at: Utc::now(),
|
|
});
|
|
|
|
let mut violations = Vec::new();
|
|
let mut warnings = Vec::new();
|
|
|
|
// Check each limit
|
|
for limit in &all_limits {
|
|
if let Some(violation) = self
|
|
.check_single_limit(limit, ¤t_exposure, request.proposed_value)
|
|
.await?
|
|
{
|
|
match violation.severity {
|
|
BreachSeverity::Warning => warnings.push(violation),
|
|
BreachSeverity::Soft | BreachSeverity::Hard | BreachSeverity::Critical => {
|
|
violations.push(violation)
|
|
},
|
|
}
|
|
}
|
|
}
|
|
|
|
let allowed = violations.is_empty();
|
|
|
|
// Calculate max allowed position/value if there are violations
|
|
let (max_allowed_position, max_allowed_value) = if !allowed {
|
|
// Find the most restrictive limit
|
|
let min_limit = all_limits
|
|
.iter()
|
|
.filter(|l| l.enabled)
|
|
.map(|l| l.threshold)
|
|
.min()
|
|
.unwrap_or(Decimal::ZERO);
|
|
|
|
let max_pos = min_limit - current_exposure.current_position;
|
|
let max_val = min_limit - current_exposure.current_value;
|
|
|
|
(
|
|
Some(max_pos.max(Decimal::ZERO)),
|
|
Some(max_val.max(Decimal::ZERO)),
|
|
)
|
|
} else {
|
|
(None, None)
|
|
};
|
|
|
|
Ok(LimitCheckResult {
|
|
allowed,
|
|
violations,
|
|
warnings,
|
|
max_allowed_position,
|
|
max_allowed_value,
|
|
})
|
|
}
|
|
|
|
async fn record_breach(&self, breach: LimitBreach) -> RiskDataResult<()> {
|
|
let query = "
|
|
INSERT INTO limit_breaches (
|
|
id, limit_id, severity, breach_time, current_value, limit_threshold,
|
|
breach_percentage, scope, scope_value, symbol, portfolio_id, strategy_id,
|
|
trader_id, acknowledged, acknowledged_by, acknowledged_at, resolved,
|
|
resolved_at, resolution_note, action_taken
|
|
) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20)
|
|
";
|
|
|
|
sqlx::query(query)
|
|
.bind(breach.id)
|
|
.bind(breach.limit_id)
|
|
.bind(breach.severity)
|
|
.bind(breach.breach_time)
|
|
.bind(breach.current_value)
|
|
.bind(breach.limit_threshold)
|
|
.bind(breach.breach_percentage)
|
|
.bind(breach.scope)
|
|
.bind(&breach.scope_value)
|
|
.bind(&breach.symbol)
|
|
.bind(&breach.portfolio_id)
|
|
.bind(&breach.strategy_id)
|
|
.bind(&breach.trader_id)
|
|
.bind(breach.acknowledged)
|
|
.bind(&breach.acknowledged_by)
|
|
.bind(breach.acknowledged_at)
|
|
.bind(breach.resolved)
|
|
.bind(breach.resolved_at)
|
|
.bind(&breach.resolution_note)
|
|
.bind(&breach.action_taken)
|
|
.execute(&self.db_pool)
|
|
.await?;
|
|
|
|
// Cache critical breaches for immediate alerting
|
|
if matches!(
|
|
breach.severity,
|
|
BreachSeverity::Critical | BreachSeverity::Hard
|
|
) {
|
|
let cache_key = format!("breach:critical:{}", breach.id);
|
|
let serialized = serde_json::to_string(&breach)?;
|
|
|
|
let mut redis_conn = self.redis_conn.clone();
|
|
redis::cmd("SETEX")
|
|
.arg(&cache_key)
|
|
.arg(7_200_i32) // 2 hours TTL
|
|
.arg(&serialized)
|
|
.query_async::<()>(&mut redis_conn)
|
|
.await?;
|
|
}
|
|
|
|
error!(
|
|
"Limit breach recorded: {:?} - {} exceeds {} by {}%",
|
|
breach.severity, breach.current_value, breach.limit_threshold, breach.breach_percentage
|
|
);
|
|
|
|
Ok(())
|
|
}
|
|
|
|
async fn get_breaches(
|
|
&self,
|
|
from: DateTime<Utc>,
|
|
to: DateTime<Utc>,
|
|
severity: Option<BreachSeverity>,
|
|
) -> RiskDataResult<Vec<LimitBreach>> {
|
|
let query = if severity.is_some() {
|
|
"
|
|
SELECT * FROM limit_breaches
|
|
WHERE breach_time BETWEEN $1 AND $2 AND severity = $3
|
|
ORDER BY breach_time DESC
|
|
"
|
|
} else {
|
|
"
|
|
SELECT * FROM limit_breaches
|
|
WHERE breach_time BETWEEN $1 AND $2
|
|
ORDER BY breach_time DESC
|
|
"
|
|
};
|
|
|
|
let breaches = if let Some(sev) = severity {
|
|
sqlx::query_as::<_, LimitBreach>(query)
|
|
.bind(from)
|
|
.bind(to)
|
|
.bind(sev)
|
|
.fetch_all(&self.db_pool)
|
|
.await?
|
|
} else {
|
|
sqlx::query_as::<_, LimitBreach>(query)
|
|
.bind(from)
|
|
.bind(to)
|
|
.fetch_all(&self.db_pool)
|
|
.await?
|
|
};
|
|
|
|
Ok(breaches)
|
|
}
|
|
|
|
async fn acknowledge_breach(
|
|
&self,
|
|
breach_id: Uuid,
|
|
acknowledged_by: String,
|
|
) -> RiskDataResult<()> {
|
|
let query = "
|
|
UPDATE limit_breaches
|
|
SET acknowledged = true, acknowledged_by = $2, acknowledged_at = $3
|
|
WHERE id = $1
|
|
";
|
|
|
|
sqlx::query(query)
|
|
.bind(breach_id)
|
|
.bind(&acknowledged_by)
|
|
.bind(Utc::now())
|
|
.execute(&self.db_pool)
|
|
.await?;
|
|
|
|
info!(
|
|
"Limit breach {} acknowledged by {}",
|
|
breach_id, acknowledged_by
|
|
);
|
|
Ok(())
|
|
}
|
|
|
|
async fn resolve_breach(
|
|
&self,
|
|
breach_id: Uuid,
|
|
resolution_note: String,
|
|
action_taken: String,
|
|
) -> RiskDataResult<()> {
|
|
let query = "
|
|
UPDATE limit_breaches
|
|
SET resolved = true, resolved_at = $2, resolution_note = $3, action_taken = $4
|
|
WHERE id = $1
|
|
";
|
|
|
|
sqlx::query(query)
|
|
.bind(breach_id)
|
|
.bind(Utc::now())
|
|
.bind(&resolution_note)
|
|
.bind(&action_taken)
|
|
.execute(&self.db_pool)
|
|
.await?;
|
|
|
|
info!("Limit breach {} resolved: {}", breach_id, resolution_note);
|
|
Ok(())
|
|
}
|
|
|
|
async fn get_utilization_summary(&self) -> RiskDataResult<HashMap<String, Decimal>> {
|
|
let query = "
|
|
SELECT
|
|
l.name,
|
|
l.threshold,
|
|
COALESCE(e.current_value, 0) as current_value
|
|
FROM position_limits l
|
|
LEFT JOIN position_exposures e ON
|
|
l.scope = e.scope AND
|
|
l.scope_value = e.scope_value
|
|
WHERE l.enabled = true
|
|
";
|
|
|
|
let rows = sqlx::query(query).fetch_all(&self.db_pool).await?;
|
|
|
|
let mut summary = HashMap::new();
|
|
|
|
for row in rows {
|
|
let name: String = row.get("name");
|
|
let threshold: Decimal = row.get("threshold");
|
|
let current: Decimal = row.get("current_value");
|
|
|
|
#[allow(clippy::arithmetic_side_effects)]
|
|
let utilization = if threshold > Decimal::ZERO {
|
|
(current / threshold) * Decimal::from(100_i32)
|
|
} else {
|
|
Decimal::ZERO
|
|
};
|
|
|
|
summary.insert(name, utilization);
|
|
}
|
|
|
|
Ok(summary)
|
|
}
|
|
|
|
async fn monitor_limits(&self) -> RiskDataResult<Vec<LimitBreach>> {
|
|
info!("Starting real-time limit monitoring");
|
|
|
|
// Get all enabled limits
|
|
let query = "SELECT * FROM position_limits WHERE enabled = true";
|
|
let limits = sqlx::query_as::<_, PositionLimit>(query)
|
|
.fetch_all(&self.db_pool)
|
|
.await?;
|
|
|
|
let mut new_breaches = Vec::new();
|
|
|
|
for limit in limits {
|
|
let scope_value = limit.scope_value.as_deref().unwrap_or("global");
|
|
|
|
if let Some(exposure) = self.get_exposure(limit.scope, scope_value).await? {
|
|
let test_value = match limit.limit_type {
|
|
LimitType::MaxPosition => exposure.current_position.abs(),
|
|
LimitType::MaxDailyVolume => exposure.daily_volume,
|
|
LimitType::MaxDailyLoss => exposure.daily_pnl.abs(),
|
|
LimitType::ExposureLimit => exposure.current_value.abs(),
|
|
LimitType::MaxDrawdown
|
|
| LimitType::VarLimit
|
|
| LimitType::ConcentrationLimit
|
|
| LimitType::LeverageLimit => {
|
|
tracing::error!(
|
|
"Unknown limit type in violation check for {:?} - using current value",
|
|
exposure.symbol
|
|
);
|
|
exposure.current_value // Conservative fallback
|
|
},
|
|
};
|
|
|
|
if test_value > limit.threshold {
|
|
#[allow(clippy::arithmetic_side_effects)]
|
|
let breach_percentage = (test_value / limit.threshold) * Decimal::from(100_i32);
|
|
let severity = Self::calculate_breach_severity(breach_percentage);
|
|
|
|
let breach = LimitBreach {
|
|
id: Uuid::new_v4(),
|
|
limit_id: limit.id,
|
|
severity,
|
|
breach_time: Utc::now(),
|
|
current_value: test_value,
|
|
limit_threshold: limit.threshold,
|
|
breach_percentage,
|
|
scope: limit.scope,
|
|
scope_value: limit.scope_value.clone(),
|
|
symbol: exposure.symbol,
|
|
portfolio_id: None, // Would be derived from scope_value
|
|
strategy_id: None,
|
|
trader_id: None,
|
|
acknowledged: false,
|
|
acknowledged_by: None,
|
|
acknowledged_at: None,
|
|
resolved: false,
|
|
resolved_at: None,
|
|
resolution_note: None,
|
|
action_taken: None,
|
|
};
|
|
|
|
new_breaches.push(breach);
|
|
}
|
|
}
|
|
}
|
|
|
|
// Record any new breaches
|
|
for breach in &new_breaches {
|
|
self.record_breach(breach.clone()).await?;
|
|
}
|
|
|
|
if !new_breaches.is_empty() {
|
|
warn!("Detected {} new limit breaches", new_breaches.len());
|
|
}
|
|
|
|
Ok(new_breaches)
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
#[allow(clippy::unwrap_used, clippy::expect_used)]
|
|
#[allow(clippy::default_numeric_fallback)]
|
|
#[allow(clippy::diverging_sub_expression)]
|
|
#[allow(clippy::used_underscore_binding)]
|
|
mod tests {
|
|
use super::*;
|
|
|
|
#[test]
|
|
fn test_breach_severity_calculation() {
|
|
// Test breach severity calculation logic inline without database
|
|
let test_cases = vec![
|
|
(Decimal::from(85), BreachSeverity::Warning), // 85% utilization
|
|
(Decimal::from(95), BreachSeverity::Soft), // 95% utilization
|
|
(Decimal::from(105), BreachSeverity::Hard), // 105% utilization
|
|
(Decimal::from(125), BreachSeverity::Critical), // 125% utilization
|
|
];
|
|
|
|
for (utilization, expected_severity) in test_cases {
|
|
let severity = if utilization >= Decimal::from(120) {
|
|
BreachSeverity::Critical
|
|
} else if utilization >= Decimal::from(100) {
|
|
BreachSeverity::Hard
|
|
} else if utilization >= Decimal::from(90) {
|
|
BreachSeverity::Soft
|
|
} else {
|
|
BreachSeverity::Warning
|
|
};
|
|
|
|
assert_eq!(
|
|
severity, expected_severity,
|
|
"Utilization {} should result in {:?}",
|
|
utilization, expected_severity
|
|
);
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn test_limit_validation() {
|
|
// Test limit validation logic inline without database
|
|
let valid_limit = PositionLimit {
|
|
id: Uuid::new_v4(),
|
|
name: "Test Limit".to_string(),
|
|
limit_type: LimitType::MaxPosition,
|
|
scope: LimitScope::Global,
|
|
scope_value: None,
|
|
threshold: Decimal::from(100000),
|
|
warning_threshold: Some(Decimal::from(80000)),
|
|
currency: "USD".to_string(),
|
|
enabled: true,
|
|
created_at: Utc::now(),
|
|
updated_at: Utc::now(),
|
|
created_by: "test_user".to_string(),
|
|
description: Some("Test limit".to_string()),
|
|
metadata: serde_json::json!({}),
|
|
};
|
|
|
|
// Inline validation logic
|
|
let is_valid = valid_limit.threshold > Decimal::ZERO
|
|
&& !valid_limit.name.is_empty()
|
|
&& !valid_limit.currency.is_empty();
|
|
assert!(is_valid, "Valid limit should pass validation");
|
|
|
|
// Test invalid limit (negative threshold)
|
|
let invalid_threshold = Decimal::from(-100);
|
|
assert!(
|
|
invalid_threshold < Decimal::ZERO,
|
|
"Negative threshold should be invalid"
|
|
);
|
|
}
|
|
}
|