diff --git a/rust_hft/apps/live/src/helpers/sentinel.rs b/rust_hft/apps/live/src/helpers/sentinel.rs index d0cf01f63..2f8ec2701 100644 --- a/rust_hft/apps/live/src/helpers/sentinel.rs +++ b/rust_hft/apps/live/src/helpers/sentinel.rs @@ -101,16 +101,10 @@ async fn run_sentinel_loop( .orders_submitted .saturating_sub(last_orders_submitted) as f64 / (check_interval_ms.max(1) as f64 / 1_000.0), - // ponytail: engine does not expose venue reconnects; wire adapter metrics here - // if Sentinel policy begins to act on reconnect frequency. - ws_reconnect_count: 0, - data_gap_count: u32::try_from( - engine_stats - .market_events_dropped - .saturating_add(engine_stats.snapshot_publish_failed) - .saturating_add(engine_stats.data_integrity_gaps), - ) - .unwrap_or(u32::MAX), + data_gap_count: engine_stats + .market_events_dropped + .saturating_add(engine_stats.snapshot_publish_failed) + .saturating_add(engine_stats.data_integrity_gaps), }; last_orders_submitted = engine_stats.orders_submitted; diff --git a/rust_hft/execution-gateway/adapters/adapter-ondo-perps/src/lib.rs b/rust_hft/execution-gateway/adapters/adapter-ondo-perps/src/lib.rs index 962c41b24..f031b9162 100644 --- a/rust_hft/execution-gateway/adapters/adapter-ondo-perps/src/lib.rs +++ b/rust_hft/execution-gateway/adapters/adapter-ondo-perps/src/lib.rs @@ -558,6 +558,7 @@ mod tests { jurisdiction: Some("SG".into()), eligibility_confirmed: true, allow_tokenized_securities: true, + ..Default::default() }, side: Side::Buy, quantity: Quantity(Decimal::from(2)), diff --git a/rust_hft/market-core/core/src/types.rs b/rust_hft/market-core/core/src/types.rs index 3f4d1c123..4611419ac 100644 --- a/rust_hft/market-core/core/src/types.rs +++ b/rust_hft/market-core/core/src/types.rs @@ -421,6 +421,24 @@ pub struct ComplianceContext { pub eligibility_confirmed: bool, #[serde(default)] pub allow_tokenized_securities: bool, + /// Executable top-of-book depth captured with the order decision. + #[serde(default)] + pub top_depth_usd: Option, + /// Spread captured with the order decision, in basis points. + #[serde(default)] + pub spread_bps: Option, + /// Corporate-action state from an authoritative reference-data source. + #[serde(default)] + pub corporate_action_active: Option, + /// Identity of the authoritative compliance/market-quality evidence producer. + #[serde(default)] + pub evidence_source: Option, + /// Venue for which the evidence was produced. + #[serde(default)] + pub evidence_venue: Option, + /// Local observation timestamp of the evidence, in microseconds since epoch. + #[serde(default)] + pub evidence_observed_at: Option, } #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] diff --git a/rust_hft/market-core/engine/tests/tokenized_security_execution_e2e.rs b/rust_hft/market-core/engine/tests/tokenized_security_execution_e2e.rs index 470e0eac9..0e11ffcc9 100644 --- a/rust_hft/market-core/engine/tests/tokenized_security_execution_e2e.rs +++ b/rust_hft/market-core/engine/tests/tokenized_security_execution_e2e.rs @@ -8,9 +8,10 @@ use engine::{ use futures::stream; use hft_core::{ AssetClass, ComplianceContext, HftResult, OrderId, OrderType, Price, ProductType, Quantity, - RegulatoryProfile, Side, Symbol, TimeInForce, + RegulatoryProfile, Side, Symbol, TimeInForce, VenueId, }; use ports::{BoxStream, ConnectionHealth, ExecutionClient, ExecutionEvent, OpenOrder, OrderIntent}; +use rust_decimal::Decimal; use tokio::sync::Mutex; #[derive(Default)] @@ -95,6 +96,12 @@ fn tokenized_security_intent() -> OrderIntent { jurisdiction: Some("AE".to_string()), eligibility_confirmed: true, allow_tokenized_securities: true, + top_depth_usd: Some(Decimal::from(100_000)), + spread_bps: Some(Decimal::ONE), + corporate_action_active: Some(false), + evidence_source: Some("paper-reference-feed".to_string()), + evidence_venue: Some(VenueId::BINANCE_TOKENIZED_SECURITIES), + evidence_observed_at: Some(hft_core::now_micros()), }) } diff --git a/rust_hft/market-core/ports/src/traits.rs b/rust_hft/market-core/ports/src/traits.rs index 09fc0df8d..7be4c3c63 100644 --- a/rust_hft/market-core/ports/src/traits.rs +++ b/rust_hft/market-core/ports/src/traits.rs @@ -560,8 +560,11 @@ pub trait RiskManager: Send + Sync { account: &AccountView, venue_specs: &std::collections::HashMap, ) -> Vec { - // 默認實現：批量調用 review()，根據 target_venue 或 symbol 查找對應 VenueSpec + // Keep a projected account across the whole batch. Calling review() with the same + // account snapshot for every intent lets individually-valid orders exceed aggregate + // position/notional limits when they are emitted in one engine tick. let mut approved_intents = Vec::new(); + let mut projected_account = account.clone(); for intent in intents { if matches!(intent.asset_class, AssetClass::TokenizedSecurity) @@ -589,7 +592,27 @@ pub trait RiskManager: Send + Sync { }; if let Some(spec) = venue_spec { - let reviewed = self.review(vec![intent], account, spec); + let reviewed = self.review(vec![intent], &projected_account, spec); + for approved in &reviewed { + let signed_quantity = match approved.side { + Side::Buy => approved.quantity.0, + Side::Sell => -approved.quantity.0, + }; + let position = projected_account + .positions + .entry(approved.symbol.clone()) + .or_insert_with(|| Position { + symbol: approved.symbol.clone(), + quantity: Quantity::zero(), + avg_price: approved.price.unwrap_or_else(Price::zero), + unrealized_pnl: rust_decimal::Decimal::ZERO, + realized_pnl: rust_decimal::Decimal::ZERO, + }); + position.quantity.0 += signed_quantity; + if let Some(price) = approved.price { + position.avg_price = price; + } + } approved_intents.extend(reviewed); } else { // 沒有找到對應的 VenueSpec，拒絕此訂單 diff --git a/rust_hft/market-core/runtime/src/exposure_projection.rs b/rust_hft/market-core/runtime/src/exposure_projection.rs new file mode 100644 index 000000000..56bf02775 --- /dev/null +++ b/rust_hft/market-core/runtime/src/exposure_projection.rs @@ -0,0 +1,181 @@ +use std::collections::HashMap; + +use hft_core::{ProductType, Symbol, VenueId}; +use ports::{AccountView, OrderIntent}; +use rust_decimal::Decimal; + +#[derive(Debug, Clone, PartialEq, Eq, Hash)] +struct ExposureKey { + venue: Option, + product_type: ProductType, + symbol: Symbol, +} + +#[derive(Debug, Clone)] +pub(crate) struct ExposureProjection { + pub symbol_gross_quantity: Decimal, + pub gross_notional: Decimal, +} + +/// Conservative projector for a batch of intents. Existing positions do not carry venue/product +/// attribution, so they cannot be proven reducible by an intent for a particular venue. They are +/// therefore retained as gross exposure; new exposure is isolated by venue + product + symbol. +#[derive(Debug, Clone)] +pub(crate) struct ExposureProjector { + remaining_quantity: HashMap, + keyed_quantity: HashMap, + keyed_notional: HashMap, + gross_notional: Decimal, +} + +impl ExposureProjector { + pub(crate) fn new(account: &AccountView) -> Self { + let remaining_quantity = account + .positions + .iter() + .map(|(symbol, position)| (symbol.clone(), position.quantity.0)) + .collect(); + let remaining_notional: HashMap<_, _> = account + .positions + .iter() + .map(|(symbol, position)| { + ( + symbol.clone(), + (position.avg_price.0 * position.quantity.0 + position.unrealized_pnl).abs(), + ) + }) + .collect(); + let gross_notional = remaining_notional.values().copied().sum(); + Self { + remaining_quantity, + keyed_quantity: HashMap::new(), + keyed_notional: HashMap::new(), + gross_notional, + } + } + + pub(crate) fn project( + &mut self, + intent: &OrderIntent, + ) -> Result { + let price = intent + .price + .map(|price| price.0) + .filter(|price| *price > Decimal::ZERO) + .ok_or("projected exposure requires a positive executable price")?; + let incoming = match intent.side { + hft_core::Side::Buy => intent.quantity.0, + hft_core::Side::Sell => -intent.quantity.0, + }; + if incoming.is_zero() { + return Err("projected exposure requires non-zero quantity"); + } + + if !incoming.is_zero() { + let key = ExposureKey { + venue: intent.target_venue, + product_type: intent.product_type, + symbol: intent.symbol.clone(), + }; + let old_quantity = self.keyed_quantity.get(&key).copied().unwrap_or_default(); + let next_quantity = old_quantity + incoming; + let old_notional = self.keyed_notional.get(&key).copied().unwrap_or_default(); + let next_notional = old_notional + incoming.abs() * price; + self.gross_notional += next_notional - old_notional; + self.keyed_quantity.insert(key.clone(), next_quantity); + self.keyed_notional.insert(key, next_notional); + } + + let unattributed = self + .remaining_quantity + .get(&intent.symbol) + .copied() + .unwrap_or_default() + .abs(); + let keyed = self + .keyed_quantity + .iter() + .filter(|(key, _)| key.symbol == intent.symbol) + .map(|(_, quantity)| quantity.abs()) + .sum::(); + Ok(ExposureProjection { + symbol_gross_quantity: unattributed + keyed, + gross_notional: self.gross_notional, + }) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use hft_core::{OrderType, Price, Quantity, Side, TimeInForce}; + + fn intent(venue: VenueId, side: Side, quantity: i64) -> OrderIntent { + OrderIntent::crypto_spot( + Symbol::new("BTCUSDT"), + side, + Quantity(Decimal::from(quantity)), + OrderType::Limit, + Some(Price(Decimal::ONE)), + TimeInForce::GTC, + "alpha".to_string(), + Some(venue), + ) + } + + #[test] + fn opposite_orders_on_different_venues_do_not_net() { + let mut projector = ExposureProjector::new(&AccountView::default()); + projector + .project(&intent(VenueId::BINANCE, Side::Buy, 60)) + .unwrap(); + let projected = projector + .project(&intent(VenueId::BITGET, Side::Sell, 60)) + .unwrap(); + + assert_eq!(projected.symbol_gross_quantity, Decimal::from(120)); + assert_eq!(projected.gross_notional, Decimal::from(120)); + } + + #[test] + fn unattributed_existing_position_cannot_be_net_reduced() { + let symbol = Symbol::new("BTCUSDT"); + let mut account = AccountView::default(); + account.positions.insert( + symbol.clone(), + ports::Position { + symbol, + quantity: Quantity(Decimal::from(60)), + avg_price: Price(Decimal::ONE), + unrealized_pnl: Decimal::ZERO, + realized_pnl: Decimal::ZERO, + }, + ); + let mut projector = ExposureProjector::new(&account); + + let projected = projector + .project(&intent(VenueId::BITGET, Side::Sell, 60)) + .unwrap(); + + assert_eq!(projected.symbol_gross_quantity, Decimal::from(120)); + assert_eq!(projected.gross_notional, Decimal::from(120)); + } + + #[test] + fn later_low_price_does_not_revalue_prior_exposure_below_the_cap() { + let mut projector = ExposureProjector::new(&AccountView::default()); + let mut expensive = intent(VenueId::BINANCE, Side::Buy, 100); + expensive.price = Some(Price(Decimal::from(1_000))); + let mut cheap = intent(VenueId::BINANCE, Side::Buy, 100); + cheap.price = Some(Price(Decimal::ONE)); + + assert_eq!( + projector.project(&expensive).unwrap().gross_notional, + Decimal::from(100_000) + ); + assert_eq!( + projector.project(&cheap).unwrap().gross_notional, + Decimal::from(100_100) + ); + } +} diff --git a/rust_hft/market-core/runtime/src/lib.rs b/rust_hft/market-core/runtime/src/lib.rs index 56a25370d..0f67d9f7b 100644 --- a/rust_hft/market-core/runtime/src/lib.rs +++ b/rust_hft/market-core/runtime/src/lib.rs @@ -8,6 +8,7 @@ //! - Risk managers //! - Event consumers +mod exposure_projection; pub mod ipc_handler; pub mod portfolio_manager; pub mod risk_manager_factory; diff --git a/rust_hft/market-core/runtime/src/portfolio_manager.rs b/rust_hft/market-core/runtime/src/portfolio_manager.rs index 8450d2e54..ebf3b1ffe 100644 --- a/rust_hft/market-core/runtime/src/portfolio_manager.rs +++ b/rust_hft/market-core/runtime/src/portfolio_manager.rs @@ -1,7 +1,11 @@ use std::collections::HashMap; +use hft_core::VenueId; +use ports::RiskManager; +use rust_decimal::Decimal; use tracing::warn; +use crate::exposure_projection::ExposureProjector; use crate::system_builder::{PortfolioSpec, StrategyConfig}; /// 聚合策略與組合設定的管理器 @@ -53,3 +57,206 @@ impl PortfolioManager { .unwrap_or_default() } } + +/// Applies configured cross-strategy portfolio budgets before the venue/global risk manager. +/// Until positions carry strategy attribution, existing account exposure is conservatively +/// charged to every portfolio instead of assuming that un-attributed exposure is harmless. +pub struct PortfolioBudgetRiskManager { + base_risk_manager: Box, + manager: PortfolioManager, +} + +impl PortfolioBudgetRiskManager { + pub fn new( + base_risk_manager: Box, + definitions: Vec, + strategies: &[StrategyConfig], + ) -> Self { + Self { + base_risk_manager, + manager: PortfolioManager::new(definitions, strategies), + } + } + + fn filter( + &self, + intents: Vec, + account: &ports::AccountView, + ) -> Vec { + let specs: HashMap<&str, &PortfolioSpec> = self + .manager + .portfolio_specs() + .iter() + .map(|spec| (spec.name.as_str(), spec)) + .collect(); + let mut projectors: HashMap = HashMap::new(); + let mut approved = Vec::with_capacity(intents.len()); + + for intent in intents { + let portfolio_names = self.manager.portfolios_for_strategy(&intent.strategy_id); + if portfolio_names.is_empty() { + approved.push(intent); + continue; + } + + let mut projections = Vec::with_capacity(portfolio_names.len()); + let mut reject_reason = None; + + for portfolio_name in &portfolio_names { + let Some(spec) = specs.get(portfolio_name.as_str()) else { + reject_reason = Some("missing portfolio definition"); + break; + }; + let mut projector = projectors + .get(portfolio_name) + .cloned() + .unwrap_or_else(|| ExposureProjector::new(account)); + let projected = match projector.project(&intent) { + Ok(projected) => projected, + Err(reason) => { + reject_reason = Some(reason); + break; + } + }; + if spec + .max_position + .is_some_and(|limit| projected.symbol_gross_quantity > limit) + { + reject_reason = Some("portfolio position budget exceeded"); + break; + } + if spec + .max_notional + .is_some_and(|limit| projected.gross_notional > limit) + { + reject_reason = Some("portfolio notional budget exceeded"); + break; + } + projections.push((portfolio_name.clone(), projector)); + } + + if let Some(reason) = reject_reason { + warn!(strategy = %intent.strategy_id, symbol = %intent.symbol, %reason, "投资组合预算拒绝订单意图"); + continue; + } + for (portfolio_name, projector) in projections { + projectors.insert(portfolio_name, projector); + } + approved.push(intent); + } + + approved + } +} + +#[cfg(test)] +mod tests { + use super::*; + use hft_core::{OrderType, Price, Quantity, Side, Symbol, TimeInForce}; + + #[test] + fn shared_portfolio_budget_accumulates_across_strategies() { + let base = crate::RiskManagerFactory::create_risk_manager(&crate::RiskConfig { + risk_type: "Default".to_string(), + global_position_limit: Decimal::from(1_000), + global_notional_limit: Decimal::from(100_000), + max_orders_per_second: 100, + staleness_threshold_us: u64::MAX, + max_daily_loss: Decimal::from(10_000), + max_drawdown_pct: 5.0, + ..Default::default() + }); + let manager = PortfolioBudgetRiskManager::new( + base, + vec![PortfolioSpec { + name: "shared".to_string(), + strategies: vec!["alpha-a".to_string(), "alpha-b".to_string()], + max_notional: Some(Decimal::from(100)), + max_position: None, + ..Default::default() + }], + &[], + ); + let intent = |strategy: &str| { + ports::OrderIntent::crypto_spot( + Symbol::new("BTCUSDT"), + Side::Buy, + Quantity(Decimal::from(60)), + OrderType::Limit, + Some(Price(Decimal::ONE)), + TimeInForce::GTC, + strategy.to_string(), + Some(VenueId::BINANCE), + ) + }; + + let approved = manager.filter( + vec![intent("alpha-a"), intent("alpha-b")], + &ports::AccountView::default(), + ); + + assert_eq!(approved.len(), 1); + } +} + +impl RiskManager for PortfolioBudgetRiskManager { + fn review_orders( + &mut self, + intents: Vec, + account: &ports::AccountView, + venue_specs: &HashMap, + ) -> Vec { + let filtered = self.filter(intents, account); + self.base_risk_manager + .review_orders(filtered, account, venue_specs) + } + + fn review( + &mut self, + intents: Vec, + account: &ports::AccountView, + venue: &ports::VenueSpec, + ) -> Vec { + let filtered = self.filter(intents, account); + self.base_risk_manager.review(filtered, account, venue) + } + + fn review_with_venue_specs( + &mut self, + intents: Vec, + account: &ports::AccountView, + venue_specs: &HashMap, + ) -> Vec { + let filtered = self.filter(intents, account); + self.base_risk_manager + .review_with_venue_specs(filtered, account, venue_specs) + } + + fn on_execution_event(&mut self, event: &ports::ExecutionEvent) { + self.base_risk_manager.on_execution_event(event) + } + + fn emergency_stop(&mut self) -> Result<(), hft_core::HftError> { + self.base_risk_manager.emergency_stop() + } + + fn get_risk_metrics(&self) -> HashMap { + self.base_risk_manager.get_risk_metrics() + } + + fn should_halt_trading(&self, account: &ports::AccountView) -> bool { + self.base_risk_manager.should_halt_trading(account) + } + + fn risk_metrics(&self) -> ports::RiskMetrics { + self.base_risk_manager.risk_metrics() + } + + fn update_config(&mut self, update: ports::RiskConfigUpdate) -> Result<(), hft_core::HftError> { + self.base_risk_manager.update_config(update) + } + + fn get_config_snapshot(&self) -> ports::RiskConfigSnapshot { + self.base_risk_manager.get_config_snapshot() + } +} diff --git a/rust_hft/market-core/runtime/src/risk_manager_factory.rs b/rust_hft/market-core/runtime/src/risk_manager_factory.rs index 45116165a..fad2e37eb 100644 --- a/rust_hft/market-core/runtime/src/risk_manager_factory.rs +++ b/rust_hft/market-core/runtime/src/risk_manager_factory.rs @@ -4,7 +4,7 @@ //! including support for per-strategy risk overrides. use chrono::{DateTime, Utc, Weekday}; -use hft_core::Quantity; +use hft_core::{AssetClass, ProductType, Quantity, VenueId}; use ports::RiskManager; use risk::{ DefaultRiskManager, EnhancedRiskConfig, EnhancedRiskManager, RiskConfig, TradingWindow, @@ -12,7 +12,11 @@ use risk::{ use std::collections::HashMap; use tracing::{debug, info, warn}; -use crate::{RiskConfig as SystemRiskConfig, StrategyRiskOverride, TradingWindowConfig}; +use crate::exposure_projection::ExposureProjector; +use crate::{ + RiskConfig as SystemRiskConfig, StrategyRiskOverride, TokenizedSecuritiesRiskConfig, + TradingWindowConfig, +}; /// Risk Manager Factory that creates risk managers with per-strategy overrides pub struct RiskManagerFactory; @@ -80,8 +84,19 @@ impl RiskManagerFactory { system_risk_config: &SystemRiskConfig, ) -> Box { let base_risk_manager = Self::create_risk_manager(system_risk_config); + let max_position = system_risk_config + .enhanced + .as_ref() + .map(|config| config.max_position_per_symbol) + .filter(|limit| *limit > rust_decimal::Decimal::ZERO) + .unwrap_or(system_risk_config.global_position_limit); + let base_risk_manager: Box = Box::new(ProjectedExposureRiskManager::new( + base_risk_manager, + max_position, + system_risk_config.global_notional_limit, + )); - if system_risk_config.strategy_overrides.is_empty() { + let strategy_aware = if system_risk_config.strategy_overrides.is_empty() { // No overrides, return base manager base_risk_manager } else { @@ -90,7 +105,13 @@ impl RiskManagerFactory { base_risk_manager, system_risk_config.strategy_overrides.clone(), )) - } + }; + + Box::new(TokenizedSecuritiesRiskManager::new( + strategy_aware, + system_risk_config.tokenized_securities.clone(), + system_risk_config.staleness_threshold_us, + )) } /// Convert trading window configuration @@ -125,6 +146,217 @@ impl RiskManagerFactory { } } +/// Enforces batch exposure on venue + product + symbol identities before legacy account views can +/// net same-named instruments across venues. +pub struct ProjectedExposureRiskManager { + base_risk_manager: Box, + max_position_per_symbol: rust_decimal::Decimal, + max_global_notional: rust_decimal::Decimal, +} + +impl ProjectedExposureRiskManager { + fn new( + base_risk_manager: Box, + max_position_per_symbol: rust_decimal::Decimal, + max_global_notional: rust_decimal::Decimal, + ) -> Self { + Self { + base_risk_manager, + max_position_per_symbol, + max_global_notional, + } + } + + fn filter( + &self, + intents: Vec, + account: &ports::AccountView, + ) -> Vec { + let mut projector = ExposureProjector::new(account); + let mut approved = Vec::with_capacity(intents.len()); + for intent in intents { + let mut next_projector = projector.clone(); + let Ok(projected) = next_projector.project(&intent) else { + warn!(symbol = %intent.symbol, "全局敞口无法投影，拒绝订单意图"); + continue; + }; + if projected.symbol_gross_quantity > self.max_position_per_symbol + || projected.gross_notional > self.max_global_notional + { + warn!(symbol = %intent.symbol, "跨 venue projected exposure 超过全局限额"); + continue; + } + projector = next_projector; + approved.push(intent); + } + approved + } +} + +impl RiskManager for ProjectedExposureRiskManager { + fn review_orders( + &mut self, + intents: Vec, + account: &ports::AccountView, + venue_specs: &HashMap, + ) -> Vec { + let filtered = self.filter(intents, account); + self.base_risk_manager + .review_orders(filtered, account, venue_specs) + } + + fn review( + &mut self, + intents: Vec, + account: &ports::AccountView, + venue: &ports::VenueSpec, + ) -> Vec { + let filtered = self.filter(intents, account); + self.base_risk_manager.review(filtered, account, venue) + } + + fn review_with_venue_specs( + &mut self, + intents: Vec, + account: &ports::AccountView, + venue_specs: &HashMap, + ) -> Vec { + let filtered = self.filter(intents, account); + self.base_risk_manager + .review_with_venue_specs(filtered, account, venue_specs) + } + + fn on_execution_event(&mut self, event: &ports::ExecutionEvent) { + self.base_risk_manager.on_execution_event(event) + } + + fn emergency_stop(&mut self) -> Result<(), hft_core::HftError> { + self.base_risk_manager.emergency_stop() + } + + fn get_risk_metrics(&self) -> HashMap { + self.base_risk_manager.get_risk_metrics() + } + + fn should_halt_trading(&self, account: &ports::AccountView) -> bool { + self.base_risk_manager.should_halt_trading(account) + } + + fn risk_metrics(&self) -> ports::RiskMetrics { + self.base_risk_manager.risk_metrics() + } + + fn update_config(&mut self, update: ports::RiskConfigUpdate) -> Result<(), hft_core::HftError> { + self.base_risk_manager.update_config(update) + } + + fn get_config_snapshot(&self) -> ports::RiskConfigSnapshot { + self.base_risk_manager.get_config_snapshot() + } +} + +/// Fail-closed policy layer for securities-like tokens. +pub struct TokenizedSecuritiesRiskManager { + base_risk_manager: Box, +} + +impl TokenizedSecuritiesRiskManager { + pub fn new( + base_risk_manager: Box, + _config: TokenizedSecuritiesRiskConfig, + _evidence_max_age_us: u64, + ) -> Self { + Self { base_risk_manager } + } + + fn is_tokenized(intent: &ports::OrderIntent) -> bool { + matches!(intent.asset_class, AssetClass::TokenizedSecurity) + || matches!(intent.product_type, ProductType::TokenizedSecuritySpot) + } + + fn filter( + &self, + intents: Vec, + _account: &ports::AccountView, + ) -> Vec { + let mut approved = Vec::with_capacity(intents.len()); + + for intent in intents { + if !Self::is_tokenized(&intent) { + approved.push(intent); + continue; + } + // ponytail: intent-carried compliance data has no independent trust root; add a + // runtime-owned attestation provider before re-enabling tokenized execution. + warn!(symbol = %intent.symbol, "证券 token 缺少运行时持有的合规证明，拒绝意图"); + } + + approved + } +} + +impl RiskManager for TokenizedSecuritiesRiskManager { + fn review_orders( + &mut self, + intents: Vec, + account: &ports::AccountView, + venue_specs: &HashMap, + ) -> Vec { + let filtered = self.filter(intents, account); + self.base_risk_manager + .review_orders(filtered, account, venue_specs) + } + + fn review( + &mut self, + intents: Vec, + account: &ports::AccountView, + venue: &ports::VenueSpec, + ) -> Vec { + let filtered = self.filter(intents, account); + self.base_risk_manager.review(filtered, account, venue) + } + + fn review_with_venue_specs( + &mut self, + intents: Vec, + account: &ports::AccountView, + venue_specs: &HashMap, + ) -> Vec { + let filtered = self.filter(intents, account); + self.base_risk_manager + .review_with_venue_specs(filtered, account, venue_specs) + } + + fn on_execution_event(&mut self, event: &ports::ExecutionEvent) { + self.base_risk_manager.on_execution_event(event) + } + + fn emergency_stop(&mut self) -> Result<(), hft_core::HftError> { + self.base_risk_manager.emergency_stop() + } + + fn get_risk_metrics(&self) -> HashMap { + self.base_risk_manager.get_risk_metrics() + } + + fn should_halt_trading(&self, account: &ports::AccountView) -> bool { + self.base_risk_manager.should_halt_trading(account) + } + + fn risk_metrics(&self) -> ports::RiskMetrics { + self.base_risk_manager.risk_metrics() + } + + fn update_config(&mut self, update: ports::RiskConfigUpdate) -> Result<(), hft_core::HftError> { + self.base_risk_manager.update_config(update) + } + + fn get_config_snapshot(&self) -> ports::RiskConfigSnapshot { + self.base_risk_manager.get_config_snapshot() + } +} + /// Strategy-aware risk manager wrapper that applies per-strategy overrides pub struct StrategyAwareRiskManager { base_risk_manager: Box, @@ -292,6 +524,9 @@ impl RiskManager for StrategyAwareRiskManager { mod tests { use super::*; use crate::EnhancedRiskSettings; + use hft_core::{ + ComplianceContext, OrderType, Price, RegulatoryProfile, Side, Symbol, TimeInForce, + }; use rust_decimal::Decimal; use std::collections::HashMap; @@ -388,4 +623,122 @@ mod tests { assert_eq!(config_with_overrides.strategy_overrides.len(), 1); let _ = strategy_aware_manager as Box; } + + #[test] + fn tokenized_security_rejects_self_reported_or_future_dated_evidence() { + let risk_config = SystemRiskConfig { + risk_type: "Default".to_string(), + global_position_limit: Decimal::from(1000), + global_notional_limit: Decimal::from(100_000), + max_daily_trades: 100, + max_orders_per_second: 10, + staleness_threshold_us: u64::MAX, + max_daily_loss: Decimal::from(10_000), + max_drawdown_pct: 5.0, + enhanced: None, + strategy_overrides: HashMap::new(), + tokenized_securities: TokenizedSecuritiesRiskConfig { + allow_trading: true, + max_notional_per_symbol: Decimal::from(1_000), + max_asset_class_notional: Decimal::from(2_000), + min_top_depth_usd: Decimal::from(10_000), + max_spread_bps: Decimal::from(10), + freeze_on_corporate_action: true, + restricted_jurisdictions: vec!["US".to_string()], + }, + }; + let context = ComplianceContext { + regulatory_profile: RegulatoryProfile::AdgmTokenizedSecurity, + jurisdiction: Some("AE".to_string()), + eligibility_confirmed: true, + allow_tokenized_securities: true, + top_depth_usd: Some(Decimal::from(20_000)), + spread_bps: Some(Decimal::from(5)), + corporate_action_active: Some(false), + evidence_source: Some("licensed-reference-feed".to_string()), + evidence_venue: Some(VenueId::BINANCE_TOKENIZED_SECURITIES), + evidence_observed_at: Some(hft_core::now_micros()), + }; + let intent = ports::OrderIntent::crypto_spot( + Symbol::new("TSLAUSDT"), + Side::Buy, + Quantity(Decimal::from(6)), + OrderType::Limit, + Some(Price(Decimal::from(100))), + TimeInForce::GTC, + "token-alpha".to_string(), + Some(VenueId::BINANCE_TOKENIZED_SECURITIES), + ) + .tokenized_security_spot(context.clone()); + let specs = HashMap::from([( + VenueId::BINANCE_TOKENIZED_SECURITIES, + ports::VenueSpec::binance_spot(), + )]); + let mut manager = RiskManagerFactory::create_strategy_aware_risk_manager(&risk_config); + + let approved = + manager.review_with_venue_specs(vec![intent], &ports::AccountView::default(), &specs); + + assert!(approved.is_empty()); + + let mut future_dated = ports::OrderIntent::crypto_spot( + Symbol::new("TSLAUSDT"), + Side::Buy, + Quantity(Decimal::from(1)), + OrderType::Limit, + Some(Price(Decimal::from(100))), + TimeInForce::GTC, + "token-alpha".to_string(), + Some(VenueId::BINANCE_TOKENIZED_SECURITIES), + ) + .tokenized_security_spot(context); + future_dated.compliance_context.evidence_observed_at = + Some(hft_core::now_micros().saturating_add(1_000_000)); + let mut manager = RiskManagerFactory::create_strategy_aware_risk_manager(&risk_config); + assert!(manager + .review_with_venue_specs(vec![future_dated], &ports::AccountView::default(), &specs,) + .is_empty()); + } + + #[test] + fn projected_exposure_does_not_net_opposite_orders_across_venues() { + let config = SystemRiskConfig { + risk_type: "Default".to_string(), + global_position_limit: Decimal::from(100), + global_notional_limit: Decimal::from(1_000), + max_orders_per_second: 100, + staleness_threshold_us: u64::MAX, + max_daily_loss: Decimal::from(10_000), + max_drawdown_pct: 5.0, + ..Default::default() + }; + let base = RiskManagerFactory::create_risk_manager(&config); + let manager = ProjectedExposureRiskManager::new( + base, + config.global_position_limit, + config.global_notional_limit, + ); + let intent = |venue, side| { + ports::OrderIntent::crypto_spot( + Symbol::new("BTCUSDT"), + side, + Quantity(Decimal::from(60)), + OrderType::Limit, + Some(Price(Decimal::ONE)), + TimeInForce::GTC, + "cross-venue".to_string(), + Some(venue), + ) + }; + + let approved = manager.filter( + vec![ + intent(VenueId::BINANCE, Side::Buy), + intent(VenueId::BITGET, Side::Sell), + ], + &ports::AccountView::default(), + ); + + assert_eq!(approved.len(), 1); + } } diff --git a/rust_hft/market-core/runtime/src/system_builder.rs b/rust_hft/market-core/runtime/src/system_builder.rs index 05f9500bd..3682f4222 100644 --- a/rust_hft/market-core/runtime/src/system_builder.rs +++ b/rust_hft/market-core/runtime/src/system_builder.rs @@ -663,6 +663,15 @@ impl SystemBuilder { // Create configurable risk manager using factory let risk_manager = crate::RiskManagerFactory::create_strategy_aware_risk_manager(&self.config.risk); + let risk_manager: Box = if self.config.portfolios.is_empty() { + risk_manager + } else { + Box::new(crate::PortfolioBudgetRiskManager::new( + risk_manager, + self.config.portfolios.clone(), + &self.config.strategies, + )) + }; engine.register_risk_manager_boxed(risk_manager); info!( "已注册风控管理器 (类型: {}, 策略覆盖数: {})", diff --git a/rust_hft/risk-control/risk/src/default_risk_manager.rs b/rust_hft/risk-control/risk/src/default_risk_manager.rs index 9f59c61d4..563ac2fc1 100644 --- a/rust_hft/risk-control/risk/src/default_risk_manager.rs +++ b/rust_hft/risk-control/risk/src/default_risk_manager.rs @@ -176,10 +176,7 @@ impl DefaultRiskManager { if position.quantity.0 == Decimal::ZERO { None } else { - Some( - position.avg_price.0 - + position.unrealized_pnl / position.quantity.0, - ) + Some(position.avg_price.0 + position.unrealized_pnl / position.quantity.0) } }) }) @@ -192,8 +189,7 @@ impl DefaultRiskManager { (position.avg_price.0 * position.quantity.0 + position.unrealized_pnl).abs() }) .sum::(); - let projected_notional = - other_notional + reference_price.abs() * new_position.0.abs(); + let projected_notional = other_notional + reference_price.abs() * new_position.0.abs(); if projected_notional > self.config.max_global_notional { return Err(format!( @@ -260,6 +256,7 @@ impl RiskManager for DefaultRiskManager { self.last_account = Some(account.clone()); self.last_account_update_us = Self::current_time_us(); let mut approved_intents = Vec::new(); + let mut projected_account = account.clone(); for mut intent in intents { // 第一步：精度标准化 @@ -285,7 +282,7 @@ impl RiskManager for DefaultRiskManager { let checks = vec![ venue_check, rate_check, - self.check_position_limits(&intent, account), + self.check_position_limits(&intent, &projected_account), self.check_daily_loss(account), self.check_drawdown(account), ]; @@ -307,6 +304,24 @@ impl RiskManager for DefaultRiskManager { ); self.update_state(&intent.symbol); + let signed_quantity = match intent.side { + hft_core::Side::Buy => intent.quantity.0, + hft_core::Side::Sell => -intent.quantity.0, + }; + let position = projected_account + .positions + .entry(intent.symbol.clone()) + .or_insert_with(|| ports::Position { + symbol: intent.symbol.clone(), + quantity: Quantity::zero(), + avg_price: intent.price.unwrap_or(Price(Decimal::ZERO)), + unrealized_pnl: Decimal::ZERO, + realized_pnl: Decimal::ZERO, + }); + position.quantity.0 += signed_quantity; + if let Some(price) = intent.price { + position.avg_price = price; + } approved_intents.push(intent); } else { // 风控拒绝 @@ -614,6 +629,7 @@ mod tests { quantity: Quantity::from_f64(0.5).unwrap(), avg_price: Price::from_f64(67000.0).unwrap(), unrealized_pnl: Decimal::from(100), + realized_pnl: Decimal::ZERO, }, ); @@ -739,6 +755,28 @@ mod tests { assert!(result.unwrap_err().contains("全局名义价值限额")); } + #[test] + fn batch_review_uses_projected_account_exposure() { + let config = RiskConfig { + max_global_notional: Decimal::from(100), + max_position_per_symbol: Quantity::from_f64(100.0).unwrap(), + aggressive_mode: true, + ..Default::default() + }; + let mut mgr = DefaultRiskManager::new(config); + let account = AccountView::default(); + let venue = VenueSpec { + min_notional: Decimal::ZERO, + ..VenueSpec::default() + }; + let first = create_test_intent("BTCUSDT", Side::Buy, 0.6, Some(100.0)); + let second = create_test_intent("BTCUSDT", Side::Buy, 0.6, Some(100.0)); + + let approved = mgr.review(vec![first, second], &account, &venue); + + assert_eq!(approved.len(), 1); + } + #[test] fn reduce_and_close_orders_bypass_an_already_breached_global_cap() { let config = RiskConfig { @@ -756,6 +794,7 @@ mod tests { quantity: Quantity::from_f64(11.0).unwrap(), avg_price: Price::from_f64(100.0).unwrap(), unrealized_pnl: Decimal::ZERO, + realized_pnl: Decimal::ZERO, }, ); @@ -766,8 +805,7 @@ mod tests { assert!(mgr.check_position_limits(&close, &account).is_ok()); assert!(mgr.check_position_limits(&increase, &account).is_err()); - account.positions.get_mut(&symbol).unwrap().quantity = - Quantity::from_f64(-11.0).unwrap(); + account.positions.get_mut(&symbol).unwrap().quantity = Quantity::from_f64(-11.0).unwrap(); let cover = create_test_intent("BTCUSDT", Side::Buy, 1.0, Some(100.0)); assert!(mgr.check_position_limits(&cover, &account).is_ok()); } @@ -860,7 +898,7 @@ mod tests { let result = PrecisionNormalizer::validate_order_minimums( &intent, Quantity::from_f64(0.0001).unwrap(), // qty check passes - Decimal::from(10), // notional check fails + Decimal::from(10), // notional check fails ); assert!(result.is_err()); assert!(result.unwrap_err().contains("名义价值低于最小值")); @@ -913,7 +951,10 @@ mod tests { } let metrics = mgr.get_risk_metrics(); - assert_eq!(*metrics.get("total_orders_today").unwrap(), Decimal::from(5)); + assert_eq!( + *metrics.get("total_orders_today").unwrap(), + Decimal::from(5) + ); } #[test] diff --git a/rust_hft/risk-control/risk/src/sentinel.rs b/rust_hft/risk-control/risk/src/sentinel.rs index 3dcbe2637..df12d95b8 100644 --- a/rust_hft/risk-control/risk/src/sentinel.rs +++ b/rust_hft/risk-control/risk/src/sentinel.rs @@ -56,15 +56,15 @@ impl Default for SentinelConfig { fn default() -> Self { Self { // 延遲閾值 - latency_warn_us: 15_000, // 15ms 警告 - latency_degrade_us: 25_000, // 25ms 降頻 - latency_stop_us: 50_000, // 50ms 停止 + latency_warn_us: 15_000, // 15ms 警告 + latency_degrade_us: 25_000, // 25ms 降頻 + latency_stop_us: 50_000, // 50ms 停止 // 回撤閾值 - drawdown_warn_pct: 2.0, // 2% 警告 - drawdown_degrade_pct: 3.0, // 3% 降頻 - drawdown_stop_pct: 5.0, // 5% 停止 - drawdown_emergency_pct: 7.0, // 7% 緊急平倉 + drawdown_warn_pct: 2.0, // 2% 警告 + drawdown_degrade_pct: 3.0, // 3% 降頻 + drawdown_stop_pct: 5.0, // 5% 停止 + drawdown_emergency_pct: 7.0, // 7% 緊急平倉 // 降頻策略 degrade_reduce_position_pct: 50.0, @@ -151,10 +151,8 @@ pub struct SystemStats { pub notional_value: f64, /// 訂單提交率 (orders/second) pub order_rate: f64, - /// WebSocket 重連次數 - pub ws_reconnect_count: u32, /// 數據間隙次數 - pub data_gap_count: u32, + pub data_gap_count: u64, } /// Sentinel 哨兵 - 自動化風控核心 @@ -167,6 +165,7 @@ pub struct Sentinel { // 回撤追蹤 consecutive_drawdown_violations: u32, + last_data_gap_count: u64, // 恢復追蹤 last_violation_time: Option, @@ -187,6 +186,7 @@ impl Sentinel { state: SentinelState::Normal, consecutive_latency_violations: 0, consecutive_drawdown_violations: 0, + last_data_gap_count: 0, last_violation_time: None, degraded_since: None, total_checks: 0, @@ -229,18 +229,32 @@ impl Sentinel { // 檢查回撤 let drawdown_action = self.check_drawdown(stats); + // A newly dropped market event or failed account snapshot invalidates the trading view. + // Stop immediately and require operator-controlled recovery. + let data_action = if stats.data_gap_count > self.last_data_gap_count { + error!( + "Market-data integrity gap detected: {} -> {}", + self.last_data_gap_count, stats.data_gap_count + ); + self.last_data_gap_count = stats.data_gap_count; + SentinelAction::Stop + } else { + SentinelAction::Continue + }; + // 合併動作（取更嚴重的） - let action = latency_action.merge(drawdown_action); + let action = latency_action.merge(drawdown_action).merge(data_action); // 更新狀態 self.update_state(action); // 檢查恢復條件 if (self.state == SentinelState::Degraded || self.state == SentinelState::Recovering) - && self.should_recover(stats) { - self.recover(); - return SentinelAction::Continue; - } + && self.should_recover(stats) + { + self.recover(); + return SentinelAction::Continue; + } action } @@ -274,7 +288,10 @@ impl Sentinel { return SentinelAction::Degrade; } } else if latency >= self.config.latency_warn_us { - warn!("Latency warning: {}us >= {}us", latency, self.config.latency_warn_us); + warn!( + "Latency warning: {}us >= {}us", + latency, self.config.latency_warn_us + ); self.total_warnings += 1; return SentinelAction::Warn; } else { @@ -494,6 +511,18 @@ mod tests { assert_eq!(sentinel.state(), SentinelState::Emergency); } + #[test] + fn test_sentinel_stops_on_new_data_gap() { + let mut sentinel = Sentinel::with_defaults(); + let action = sentinel.check(&SystemStats { + data_gap_count: 1, + ..Default::default() + }); + + assert_eq!(action, SentinelAction::Stop); + assert_eq!(sentinel.state(), SentinelState::Stopped); + } + #[test] fn test_action_merge() { assert_eq!(