From a39210b4bdec35258d1f6facd85fb58abc71594c Mon Sep 17 00:00:00 2001 From: proerror Date: Thu, 16 Jul 2026 13:29:17 +0800 Subject: [PATCH 1/5] fix(engine): require authoritative account reconciliation --- rust_hft/Cargo.lock | 1 + .../adapters/adapter-polymarket/src/lib.rs | 1 + rust_hft/market-core/engine/Cargo.toml | 1 + .../engine/src/execution_control.rs | 494 ++++++++++++++++-- .../engine/src/execution_queues.rs | 1 + .../engine/src/execution_worker.rs | 235 +++++++++ rust_hft/market-core/engine/src/lib.rs | 155 +++++- rust_hft/market-core/ports/src/traits.rs | 3 + .../market-core/runtime/src/ipc_handler.rs | 5 +- .../risk-control/portfolio-core/src/lib.rs | 8 +- 10 files changed, 846 insertions(+), 58 deletions(-) diff --git a/rust_hft/Cargo.lock b/rust_hft/Cargo.lock index 85c4b2daf..780bae224 100644 --- a/rust_hft/Cargo.lock +++ b/rust_hft/Cargo.lock @@ -5289,6 +5289,7 @@ dependencies = [ "hft-data-adapter-bitget", "hft-infra-metrics", "hft-instrument", + "hft-oms-core", "hft-ports", "hft-snapshot", "hft-strategy-arbitrage", diff --git a/rust_hft/execution-gateway/adapters/adapter-polymarket/src/lib.rs b/rust_hft/execution-gateway/adapters/adapter-polymarket/src/lib.rs index 7502c9685..c3f7c1c29 100644 --- a/rust_hft/execution-gateway/adapters/adapter-polymarket/src/lib.rs +++ b/rust_hft/execution-gateway/adapters/adapter-polymarket/src/lib.rs @@ -4107,6 +4107,7 @@ impl ExecutionClient for PolymarketExecutionClient { quantity: Quantity(position.size), avg_price: Price(position.avg_price), unrealized_pnl: position.cash_pnl, + realized_pnl: position.realized_pnl, }) .collect()) } diff --git a/rust_hft/market-core/engine/Cargo.toml b/rust_hft/market-core/engine/Cargo.toml index ddb17f9a7..4fa00868a 100644 --- a/rust_hft/market-core/engine/Cargo.toml +++ b/rust_hft/market-core/engine/Cargo.toml @@ -33,6 +33,7 @@ metrics = ["dep:infra-metrics"] legacy-sdk = [] [dev-dependencies] +hft-oms-core = { path = "../../risk-control/oms-core" } hft-data-adapter-bitget = { path = "../../data-pipelines/adapters/adapter-bitget" } strategy-trend = { package = "hft-strategy-trend", path = "../../strategy-framework/strategies/trend" } strategy-arbitrage = { package = "hft-strategy-arbitrage", path = "../../strategy-framework/strategies/arbitrage" } diff --git a/rust_hft/market-core/engine/src/execution_control.rs b/rust_hft/market-core/engine/src/execution_control.rs index 64d2843f5..c7413b273 100644 --- a/rust_hft/market-core/engine/src/execution_control.rs +++ b/rust_hft/market-core/engine/src/execution_control.rs @@ -16,6 +16,7 @@ pub struct RuntimeReconciliationReport { pub order_report: OrderReconciliationReport, pub balance_report: Option, pub position_report: Option, + pub fill_report: Option, pub complete: bool, pub healthy: bool, } @@ -49,6 +50,15 @@ pub struct PositionReconciliationReport { pub healthy: bool, } +#[derive(Debug, Clone)] +pub struct FillReconciliationReport { + /// Authoritative venue fills that are absent from the local accounting ledger. + pub exchange_only_fill_ids: Vec, + pub client_errors: Vec, + pub complete: bool, + pub healthy: bool, +} + /// Shared control plane for Sentinel, IPC, and gRPC. /// /// The engine lock is held only while changing mode or taking an OMS snapshot. @@ -362,30 +372,36 @@ impl ExecutionControlHandle { &self, include_balances: bool, ) -> HftResult { + if include_balances { + self.engine + .lock() + .await + .publish_runtime_truth_status(crate::RuntimeTruthStatus { + reconciliation_complete: false, + reconciliation_healthy: false, + observed_at_us: hft_core::now_micros(), + account_id: None, + }); + } let worker_tx = self.worker_sender()?; let (reply_tx, reply_rx) = oneshot::channel(); worker_tx .send(ControlCommand::Reconcile { include_balances, include_positions: include_balances, - include_recent_fills: false, + include_recent_fills: include_balances, reply: reply_tx, }) .map_err(|_| HftError::Execution("execution worker control channel closed".into()))?; let worker_snapshot = self.await_reply(reply_rx, "reconciliation").await?; - let exchange_orders = worker_snapshot - .clients - .iter() - .filter_map(|client| client.open_orders.as_ref().ok()) - .flatten() - .cloned() - .collect::>(); let order_report = { let engine = self.engine.lock().await; - engine.reconcile_open_orders(&exchange_orders) + engine.reconcile_open_orders(&worker_snapshot) }; - let (balance_report, position_report) = if include_balances { - let account_view = self.engine.lock().await.get_account_view(); + let (balance_report, position_report, fill_report) = if include_balances { + let engine = self.engine.lock().await; + let account_view = engine.get_account_view(); + let portfolio_state = engine.export_portfolio_state(); ( Some(reconcile_balances( &worker_snapshot, @@ -393,17 +409,27 @@ impl ExecutionControlHandle { self.balance_tolerance_usd, )), reconcile_positions(&worker_snapshot, &account_view.positions), + Some(reconcile_recent_fills( + &worker_snapshot, + &portfolio_state.processed_fill_ids, + )), ) } else { - (None, None) + (None, None, None) }; let complete = worker_snapshot.is_complete() + && (!include_balances + || worker_snapshot.clients.iter().all(|client| { + client.positions.as_ref().is_some_and(Result::is_ok) + && client.recent_fills.as_ref().is_some_and(Result::is_ok) + })) && balance_report .as_ref() .is_none_or(|balances| balances.complete) && position_report .as_ref() - .is_none_or(|positions| positions.complete); + .is_none_or(|positions| positions.complete) + && fill_report.as_ref().is_none_or(|fills| fills.complete); let healthy = complete && !order_report.has_discrepancies() && balance_report @@ -411,16 +437,30 @@ impl ExecutionControlHandle { .is_none_or(|balances| balances.healthy) && position_report .as_ref() - .is_none_or(|positions| positions.healthy); + .is_none_or(|positions| positions.healthy) + && fill_report.as_ref().is_none_or(|fills| fills.healthy); - Ok(RuntimeReconciliationReport { + let report = RuntimeReconciliationReport { worker_snapshot, order_report, balance_report, position_report, + fill_report, complete, healthy, - }) + }; + if include_balances { + self.engine + .lock() + .await + .publish_runtime_truth_status(crate::RuntimeTruthStatus { + reconciliation_complete: report.complete, + reconciliation_healthy: report.healthy, + observed_at_us: hft_core::now_micros(), + account_id: report.worker_snapshot.account_id(), + }); + } + Ok(report) } /// Quiesce both strategy production and worker intake before taking an authoritative @@ -514,6 +554,9 @@ impl ExecutionControlHandle { let mut cash_balance = Decimal::ZERO; let mut positions = std::collections::HashMap::new(); + let mut baseline_processed_fill_ids = + std::collections::HashMap::>::new(); + let mut baseline_recent_accounting_event_ids = Vec::new(); let mut saw_balances = false; for client in &snapshot.clients { let balances = client.balances.as_ref().ok_or_else(|| { @@ -562,6 +605,38 @@ impl ExecutionControlHandle { ))); } } + + let recent_fills = client.recent_fills.as_ref().ok_or_else(|| { + HftError::Execution(format!( + "execution client {} did not provide recent fills for account bootstrap", + client.client_index + )) + })?; + for fill in recent_fills.as_ref().map_err(|error| { + HftError::Execution(format!( + "execution client {} recent-fill bootstrap failed: {error}", + client.client_index + )) + })? { + if fill.fill_id.is_empty() { + return Err(HftError::Execution(format!( + "execution client {} returned a recent fill without fill_id", + client.client_index + ))); + } + let inserted = baseline_processed_fill_ids + .entry(fill.order_id.clone()) + .or_default() + .insert(fill.fill_id.clone()); + if !inserted { + return Err(HftError::Execution(format!( + "account bootstrap found duplicate recent fill {}:{}", + fill.order_id.0, fill.fill_id + ))); + } + baseline_recent_accounting_event_ids + .push((fill.order_id.clone(), format!("fill:{}", fill.fill_id))); + } } if !saw_balances { return Err(HftError::Execution( @@ -599,8 +674,8 @@ impl ExecutionControlHandle { account_view, order_meta: current.order_meta, market_prices: current.market_prices, - processed_fill_ids: current.processed_fill_ids, - recent_accounting_event_ids: current.recent_accounting_event_ids, + processed_fill_ids: baseline_processed_fill_ids, + recent_accounting_event_ids: baseline_recent_accounting_event_ids, })?; Ok(true) } @@ -724,18 +799,70 @@ fn reconcile_balances( } } +fn reconcile_recent_fills( + snapshot: &WorkerReconcileSnapshot, + processed_fill_ids: &std::collections::HashMap>, +) -> FillReconciliationReport { + let mut exchange_only_fill_ids = Vec::new(); + let mut client_errors = Vec::new(); + let mut observed = std::collections::HashSet::new(); + + if snapshot.clients.is_empty() { + client_errors.push("no execution clients returned recent-fill snapshots".to_string()); + } + for client in &snapshot.clients { + match &client.recent_fills { + None => client_errors.push(format!( + "client={} does not support authoritative recent fills", + client.client_index + )), + Some(Err(error)) => client_errors.push(format!( + "client={} recent-fill snapshot failed: {}", + client.client_index, error + )), + Some(Ok(fills)) => { + for fill in fills { + let identity = (fill.order_id.clone(), fill.fill_id.clone()); + if fill.fill_id.is_empty() { + client_errors.push(format!( + "client={} returned a recent fill without fill_id", + client.client_index + )); + continue; + } + if !observed.insert(identity.clone()) { + client_errors.push(format!( + "duplicate authoritative fill identity order={} fill={}", + fill.order_id.0, fill.fill_id + )); + continue; + } + let locally_processed = processed_fill_ids + .get(&fill.order_id) + .is_some_and(|ids| ids.contains(&fill.fill_id)); + if !locally_processed { + exchange_only_fill_ids + .push(format!("{}:{}", fill.order_id.0, fill.fill_id)); + } + } + } + } + } + exchange_only_fill_ids.sort(); + let complete = client_errors.is_empty(); + let healthy = complete && exchange_only_fill_ids.is_empty(); + FillReconciliationReport { + exchange_only_fill_ids, + client_errors, + complete, + healthy, + } +} + fn reconcile_positions( snapshot: &WorkerReconcileSnapshot, local_positions: &std::collections::HashMap, ) -> Option { - if snapshot - .clients - .iter() - .all(|client| client.positions.is_none()) - { - return None; - } - #[derive(Default)] struct VenuePositions { quantities: BTreeMap, @@ -922,6 +1049,7 @@ mod tests { use super::*; use crate::{EngineConfig, TradingMode}; use hft_core::{Price, Quantity, Side}; + use oms_core::OmsCore; use ports::{ AccountBalance, ExecutionEvent, OrderManager, OrderUpdate, PortfolioManager, RegisterOrderParams, @@ -1132,6 +1260,212 @@ mod tests { worker.await.expect("worker task"); } + #[tokio::test] + async fn reconciliation_does_not_match_same_order_id_across_client_identity() { + let order_id = OrderId("42".to_string()); + let binance_account = AccountId("binance-main".to_string()); + let mut engine = Engine::new(EngineConfig::default()); + engine.set_order_manager(Box::new(ReconcileOrderManager { + order: OrderRecord { + order_id: order_id.clone(), + client_order_id: Some("client-42".to_string()), + symbol: Symbol::new("BTCUSDT"), + side: Side::Buy, + qty: Quantity(Decimal::ONE), + cum_qty: Quantity::zero(), + avg_price: Some(Price(Decimal::from(100))), + status: OrderStatus::Acknowledged, + venue: Some(VenueId::BINANCE), + strategy_id: Some("test".to_string()), + }, + })); + engine + .order_account_map + .insert(order_id.clone(), binance_account.clone()); + let engine = Arc::new(Mutex::new(engine)); + let (worker_tx, mut worker_rx) = mpsc::unbounded_channel(); + let control = ExecutionControlHandle::new(engine, Some(worker_tx), true); + let worker = tokio::spawn({ + async move { + let open_order = |exchange_order_id: &str, symbol: &str| ports::OpenOrder { + order_id: OrderId(exchange_order_id.to_string()), + client_order_id: Some("client-42".to_string()), + symbol: Symbol::new(symbol), + side: Side::Buy, + order_type: hft_core::OrderType::Limit, + original_quantity: Quantity(Decimal::ONE), + remaining_quantity: Quantity(Decimal::ONE), + filled_quantity: Quantity::zero(), + price: Some(Price(Decimal::from(100))), + status: OrderStatus::Accepted, + created_at: 1, + updated_at: 1, + }; + match worker_rx.recv().await.expect("reconcile command") { + ControlCommand::Reconcile { reply, .. } => reply + .send(WorkerReconcileSnapshot { + clients: vec![ + crate::execution_worker::ClientReconcileSnapshot { + client_index: 0, + venue: Some(VenueId::BINANCE), + account_id: Some(binance_account), + open_orders: Ok(vec![open_order("42", "BTCUSDT")]), + balances: None, + positions: None, + recent_fills: None, + }, + crate::execution_worker::ClientReconcileSnapshot { + client_index: 1, + venue: Some(VenueId::BITGET), + account_id: Some(AccountId("bitget-main".to_string())), + open_orders: Ok(vec![ + open_order("42", "ETHUSDT"), + open_order("bitget-42", "ETHUSDT"), + ]), + balances: None, + positions: None, + recent_fills: None, + }, + ], + }) + .expect("send reconciliation"), + _ => panic!("unexpected control command"), + } + } + }); + + let report = control.reconcile(false).await.expect("reconcile"); + + assert!(report.complete); + assert!(!report.healthy); + assert_eq!( + report.order_report.exchange_only, + vec![order_id, OrderId("bitget-42".to_string())] + ); + worker.await.expect("worker task"); + } + + #[tokio::test] + async fn imported_oms_state_restores_account_identity_for_reconciliation() { + let order_id = OrderId("restored-42".to_string()); + let account_id = AccountId("binance-main".to_string()); + let mut engine = Engine::new(EngineConfig::default()); + engine.set_order_manager(Box::new(OmsCore::new())); + engine.set_strategy_account_mapping(HashMap::from([( + "alpha".to_string(), + account_id.clone(), + )])); + engine.import_oms_state(HashMap::from([( + order_id.clone(), + OrderRecord { + order_id: order_id.clone(), + client_order_id: Some("client-restored-42".to_string()), + symbol: Symbol::new("BTCUSDT"), + side: Side::Buy, + qty: Quantity(Decimal::ONE), + cum_qty: Quantity::zero(), + avg_price: Some(Price(Decimal::from(100))), + status: OrderStatus::Acknowledged, + venue: Some(VenueId::BINANCE), + strategy_id: Some("alpha".to_string()), + }, + )])); + assert_eq!( + engine.get_account_for_order(&order_id), + Some(account_id.clone()) + ); + let engine = Arc::new(Mutex::new(engine)); + let (worker_tx, mut worker_rx) = mpsc::unbounded_channel(); + let control = ExecutionControlHandle::new(engine, Some(worker_tx), true); + let worker = tokio::spawn({ + let order_id = order_id.clone(); + async move { + match worker_rx.recv().await.expect("reconcile command") { + ControlCommand::Reconcile { reply, .. } => reply + .send(WorkerReconcileSnapshot { + clients: vec![crate::execution_worker::ClientReconcileSnapshot { + client_index: 0, + venue: Some(VenueId::BINANCE), + account_id: Some(account_id), + open_orders: Ok(vec![ports::OpenOrder { + order_id, + client_order_id: Some("client-restored-42".to_string()), + symbol: Symbol::new("BTCUSDT"), + side: Side::Buy, + order_type: hft_core::OrderType::Limit, + original_quantity: Quantity(Decimal::ONE), + remaining_quantity: Quantity(Decimal::ONE), + filled_quantity: Quantity::zero(), + price: Some(Price(Decimal::from(100))), + status: OrderStatus::Accepted, + created_at: 1, + updated_at: 1, + }]), + balances: None, + positions: None, + recent_fills: None, + }], + }) + .expect("send reconciliation"), + _ => panic!("unexpected control command"), + } + } + }); + + let report = control.reconcile(false).await.expect("reconcile"); + + assert!(report.complete); + assert!(report.healthy); + assert!(!report.order_report.has_discrepancies()); + worker.await.expect("worker task"); + } + + #[tokio::test] + async fn order_only_reconciliation_does_not_publish_authoritative_account_truth() { + let engine = Arc::new(Mutex::new(Engine::new(EngineConfig::default()))); + let runtime_truth = engine.lock().await.runtime_truth_reader(); + let (worker_tx, mut worker_rx) = mpsc::unbounded_channel(); + let control = ExecutionControlHandle::new(engine, Some(worker_tx), true); + let worker = tokio::spawn(async move { + match worker_rx.recv().await.expect("control command") { + ControlCommand::Reconcile { + include_balances, + include_positions, + include_recent_fills, + reply, + } => { + assert!(!include_balances); + assert!(!include_positions); + assert!(!include_recent_fills); + reply + .send(WorkerReconcileSnapshot { + clients: vec![crate::execution_worker::ClientReconcileSnapshot { + client_index: 0, + venue: Some(VenueId::POLYMARKET), + account_id: None, + open_orders: Ok(Vec::new()), + balances: None, + positions: None, + recent_fills: None, + }], + }) + .expect("send reconciliation"); + } + _ => panic!("unexpected control command"), + } + }); + + let report = control.reconcile(false).await.expect("reconcile"); + worker.await.expect("worker task"); + + assert!(report.complete); + assert!(report.healthy); + let truth = runtime_truth.load(); + assert!(!truth.reconciliation_complete); + assert!(!truth.reconciliation_healthy); + assert_eq!(truth.observed_at_us, 0); + } + #[tokio::test] async fn guarded_reconciliation_quiesces_before_snapshot_and_resumes_only_when_healthy() { let engine = Arc::new(Mutex::new(Engine::new(EngineConfig::default()))); @@ -1274,10 +1608,13 @@ mod tests { match worker_rx.recv().await.expect("control command") { ControlCommand::Reconcile { include_balances, + include_positions, + include_recent_fills, reply, - .. } => { assert!(include_balances); + assert!(include_positions); + assert!(include_recent_fills); reply .send(WorkerReconcileSnapshot { clients: vec![crate::execution_worker::ClientReconcileSnapshot { @@ -1286,8 +1623,8 @@ mod tests { account_id: None, open_orders: Ok(Vec::new()), balances: Some(balances), - positions: None, - recent_fills: None, + positions: Some(Ok(Vec::new())), + recent_fills: Some(Ok(Vec::new())), }], }) .expect("send reconciliation"); @@ -1358,6 +1695,80 @@ mod tests { assert_eq!(balances.missing_valuations, vec!["client=0 asset=BTC"]); } + #[test] + fn recent_fill_reconciliation_detects_unaccounted_exchange_fill() { + let order_id = OrderId("order-1".to_string()); + let snapshot = WorkerReconcileSnapshot { + clients: vec![crate::execution_worker::ClientReconcileSnapshot { + client_index: 0, + venue: Some(VenueId::POLYMARKET), + account_id: None, + open_orders: Ok(Vec::new()), + balances: None, + positions: None, + recent_fills: Some(Ok(vec![ports::AccountFill { + fill_id: "fill-1".to_string(), + order_id: order_id.clone(), + symbol: Symbol::new("123"), + side: Side::Buy, + price: Price(Decimal::new(5, 1)), + quantity: Quantity(Decimal::ONE), + fee: None, + timestamp: 1, + }])), + }], + }; + + let missing = reconcile_recent_fills(&snapshot, &HashMap::new()); + assert!(missing.complete); + assert!(!missing.healthy); + assert_eq!(missing.exchange_only_fill_ids, vec!["order-1:fill-1"]); + + let processed = HashMap::from([( + order_id, + std::collections::HashSet::from(["fill-1".to_string()]), + )]); + let matched = reconcile_recent_fills(&snapshot, &processed); + assert!(matched.healthy); + } + + #[tokio::test] + async fn live_reconciliation_is_incomplete_without_positions_and_recent_fills() { + let (worker_tx, mut worker_rx) = mpsc::unbounded_channel(); + let control = + ExecutionControlHandle::new(engine_with_equity(Decimal::ZERO), Some(worker_tx), true); + let worker = tokio::spawn(async move { + match worker_rx.recv().await.expect("control command") { + ControlCommand::Reconcile { reply, .. } => reply + .send(WorkerReconcileSnapshot { + clients: vec![crate::execution_worker::ClientReconcileSnapshot { + client_index: 0, + venue: Some(VenueId::BINANCE), + account_id: None, + open_orders: Ok(Vec::new()), + balances: Some(Ok(Vec::new())), + positions: None, + recent_fills: None, + }], + }) + .expect("send reconciliation"), + _ => panic!("unexpected control command"), + } + }); + + let report = control.reconcile(true).await.expect("reconcile"); + worker.await.expect("worker task"); + + assert!(!report.complete); + assert!(!report.healthy); + assert!(report + .position_report + .expect("position report") + .client_errors + .iter() + .any(|error| error.contains("does not support"))); + } + #[test] fn position_reconciliation_detects_exchange_only_and_quantity_mismatch() { let token_a = Symbol::new("111"); @@ -1369,6 +1780,7 @@ mod tests { quantity: Quantity(Decimal::from(2)), avg_price: Price(Decimal::new(45, 2)), unrealized_pnl: Decimal::ZERO, + realized_pnl: Decimal::ZERO, }, )]); let snapshot = WorkerReconcileSnapshot { @@ -1384,12 +1796,14 @@ mod tests { quantity: Quantity(Decimal::from(3)), avg_price: Price(Decimal::new(45, 2)), unrealized_pnl: Decimal::ZERO, + realized_pnl: Decimal::ZERO, }, ports::Position { symbol: token_b.clone(), quantity: Quantity(Decimal::ONE), avg_price: Price(Decimal::new(55, 2)), unrealized_pnl: Decimal::ZERO, + realized_pnl: Decimal::ZERO, }, ])), recent_fills: None, @@ -1415,6 +1829,7 @@ mod tests { quantity: Quantity(Decimal::ONE), avg_price: Price(Decimal::new(50, 2)), unrealized_pnl: Decimal::ZERO, + realized_pnl: Decimal::ZERO, }, )]); let snapshot = WorkerReconcileSnapshot { @@ -1462,6 +1877,7 @@ mod tests { quantity: Quantity(Decimal::ONE), avg_price: Price(Decimal::from(100)), unrealized_pnl: Decimal::ZERO, + realized_pnl: Decimal::ZERO, }, )]); let snapshot = WorkerReconcileSnapshot { @@ -1544,12 +1960,22 @@ mod tests { Some(Decimal::from(80)), )])), positions: Some(Ok(vec![ports::Position { - symbol: token, + symbol: token.clone(), quantity: Quantity(Decimal::from(2)), avg_price: Price(Decimal::new(45, 2)), unrealized_pnl: Decimal::new(10, 2), + realized_pnl: Decimal::ZERO, + }])), + recent_fills: Some(Ok(vec![ports::AccountFill { + fill_id: "historical-fill-1".to_string(), + order_id: OrderId("historical-order-1".to_string()), + symbol: token, + side: Side::Buy, + price: Price(Decimal::new(45, 2)), + quantity: Quantity(Decimal::from(2)), + fee: None, + timestamp: 1, }])), - recent_fills: None, }], } } @@ -1572,6 +1998,12 @@ mod tests { assert_eq!(account.unrealized_pnl, Decimal::new(10, 2)); assert_eq!(account.high_water_mark, account.equity()); assert!(account.session_start_us > 0); + drop(account); + let state = engine.lock().await.export_portfolio_state(); + assert!(state + .processed_fill_ids + .get(&OrderId("historical-order-1".to_string())) + .is_some_and(|ids| ids.contains("historical-fill-1"))); assert!(!control .bootstrap_pristine_account(&polymarket_account_snapshot(Vec::new())) diff --git a/rust_hft/market-core/engine/src/execution_queues.rs b/rust_hft/market-core/engine/src/execution_queues.rs index 8dee7e150..490688acf 100644 --- a/rust_hft/market-core/engine/src/execution_queues.rs +++ b/rust_hft/market-core/engine/src/execution_queues.rs @@ -129,6 +129,7 @@ pub fn create_execution_queues(config: ExecutionQueueConfig) -> (EngineQueues, W impl EngineQueues { /// 发送订单意图到执行 worker (非阻塞) + #[allow(clippy::result_large_err)] // Returning ownership avoids a heap allocation on the hot path. pub fn send_intent(&mut self, intent: OrderIntent) -> Result<(), OrderIntent> { let now = hft_core::now_micros(); let envelope = OrderIntentEnvelope::new( diff --git a/rust_hft/market-core/engine/src/execution_worker.rs b/rust_hft/market-core/engine/src/execution_worker.rs index 1146975b4..e75da19b2 100644 --- a/rust_hft/market-core/engine/src/execution_worker.rs +++ b/rust_hft/market-core/engine/src/execution_worker.rs @@ -569,6 +569,10 @@ impl ExecutionWorker { .record_submission_latency(execution_latency as f64); self.stats.orders_placed += 1; + if self.latch_duplicate_order_id(&order_id, client_idx) { + self.stats.orders_failed += 1; + continue; + } self.order_to_client.insert(order_id.clone(), client_idx); let venue_for_client = intent @@ -657,6 +661,9 @@ impl ExecutionWorker { .find_map(|(account_id, index)| { (*index == client_idx).then_some(account_id.clone()) }); + if self.latch_duplicate_order_id(&order_id, client_idx) { + continue; + } self.order_to_client.insert(order_id.clone(), client_idx); self.tracked_orders.insert( order_id.clone(), @@ -786,6 +793,25 @@ impl ExecutionWorker { ) } + fn latch_duplicate_order_id(&mut self, order_id: &OrderId, client_idx: usize) -> bool { + if !self.order_to_client.contains_key(order_id) + && !self.tracked_orders.contains_key(order_id) + && !self.pending_acks.contains_key(order_id) + { + return false; + } + + error!( + order_id = %order_id.0, + existing_client_idx = ?self.order_to_client.get(order_id), + duplicate_client_idx = client_idx, + "duplicate execution order id; preserving the first lifecycle and latching intake" + ); + self.accepting_intents = false; + self.emergency_latched = true; + true + } + /// 轮询执行回报流 async fn poll_execution_events(&mut self) -> u32 { let mut events_count = 0; @@ -796,6 +822,10 @@ impl ExecutionWorker { client_idx, result: Ok(event), })) => { + if !self.private_order_event_matches_client(client_idx, &event) { + events_count += 1; + continue; + } let requires_reconciliation = self.event_requires_reconciliation(client_idx, &event); self.update_connection_tracking(client_idx, &event); @@ -841,6 +871,9 @@ impl ExecutionWorker { client_idx, result: Ok(event), }) => { + if !self.private_order_event_matches_client(client_idx, &event) { + return; + } let requires_reconciliation = self.event_requires_reconciliation(client_idx, &event); self.update_connection_tracking(client_idx, &event); @@ -1004,6 +1037,39 @@ impl ExecutionWorker { } } + fn private_order_event_matches_client( + &mut self, + client_idx: usize, + event: &ExecutionEvent, + ) -> bool { + let order_id = match event { + ExecutionEvent::OrderAck { order_id, .. } + | ExecutionEvent::Fill { order_id, .. } + | ExecutionEvent::FeeCharged { order_id, .. } + | ExecutionEvent::OrderReject { order_id, .. } + | ExecutionEvent::OrderCompleted { order_id, .. } + | ExecutionEvent::OrderCanceled { order_id, .. } + | ExecutionEvent::OrderModified { order_id, .. } => order_id, + _ => return true, + }; + let Some(expected_client_idx) = self.order_to_client.get(order_id).copied() else { + return true; + }; + if expected_client_idx == client_idx { + return true; + } + + error!( + order_id = %order_id.0, + expected_client_idx, + conflicting_client_idx = client_idx, + "discarded private order event from the wrong execution client" + ); + self.emergency_latched = true; + self.latch_stream_recovery(); + false + } + fn latch_stream_recovery(&mut self) { self.accepting_intents = false; self.stream_recovery_pending = true; @@ -1604,6 +1670,14 @@ pub struct WorkerReconcileSnapshot { } impl WorkerReconcileSnapshot { + pub(crate) fn account_id(&self) -> Option { + let mut clients = self.clients.iter(); + let account_id = clients.next()?.account_id.as_ref()?; + clients + .all(|client| client.account_id.as_ref() == Some(account_id)) + .then(|| account_id.clone()) + } + pub fn is_complete(&self) -> bool { !self.clients.is_empty() && self.clients.iter().all(|client| { @@ -1878,6 +1952,37 @@ mod tests { use std::collections::HashSet; use std::sync::Mutex as StdMutex; + #[test] + fn reconciliation_account_scope_requires_one_complete_identity() { + let client = |client_index, account_id: Option<&str>| ClientReconcileSnapshot { + client_index, + venue: None, + account_id: account_id.map(|value| AccountId(value.to_string())), + open_orders: Ok(Vec::new()), + balances: None, + positions: None, + recent_fills: None, + }; + + let one_account = WorkerReconcileSnapshot { + clients: vec![client(0, Some("account-1")), client(1, Some("account-1"))], + }; + assert_eq!( + one_account.account_id(), + Some(AccountId("account-1".to_string())) + ); + assert!(WorkerReconcileSnapshot { + clients: vec![client(0, Some("account-1")), client(1, Some("account-2"))], + } + .account_id() + .is_none()); + assert!(WorkerReconcileSnapshot { + clients: vec![client(0, Some("account-1")), client(1, None)], + } + .account_id() + .is_none()); + } + #[tokio::test] async fn downstream_backpressure_prevents_prefetching_the_next_adapter_event() { let queue_config = crate::ExecutionQueueConfig { @@ -3036,6 +3141,136 @@ mod tests { )); } + #[tokio::test] + async fn duplicate_order_id_across_clients_latches_without_overwriting_first_order() { + let first_state = Arc::new(StdMutex::new(MockExecutionState::default())); + let second_state = Arc::new(StdMutex::new(MockExecutionState::default())); + let clients = vec![ + Box::new(MockExecutionClient { + state: first_state, + place_error: false, + list_error: false, + cancel_error: false, + }) as Box, + Box::new(MockExecutionClient { + state: second_state, + place_error: false, + list_error: false, + cancel_error: false, + }) as Box, + ]; + let (mut engine_queues, worker_queues) = + crate::create_execution_queues(crate::ExecutionQueueConfig::default()); + let mut first = create_test_intent("BTCUSDT"); + first.strategy_id = "binance".to_string(); + first.target_venue = Some(VenueId::BINANCE); + let mut second = create_test_intent("ETHUSDT"); + second.strategy_id = "bitget".to_string(); + second.target_venue = Some(VenueId::BITGET); + engine_queues + .send_intent(first) + .expect("queue first intent"); + engine_queues + .send_intent(second) + .expect("queue second intent"); + let (_control_tx, control_rx) = mpsc::unbounded_channel(); + let mut worker = ExecutionWorker::new( + ExecutionWorkerConfig::default(), + worker_queues, + clients, + control_rx, + ); + worker.venue_to_client = [(VenueId::BINANCE, 0usize), (VenueId::BITGET, 1usize)] + .into_iter() + .collect(); + worker.account_to_client = [ + (AccountId("binance-main".to_string()), 0usize), + (AccountId("bitget-main".to_string()), 1usize), + ] + .into_iter() + .collect(); + let mut queued = worker.queues.receive_envelopes(); + + worker.process_order_intents(&mut queued).await; + + let order_id = OrderId("placed".to_string()); + assert!(!worker.accepting_intents); + assert!(worker.emergency_latched); + assert_eq!(worker.order_to_client.get(&order_id), Some(&0)); + assert_eq!( + worker + .tracked_orders + .get(&order_id) + .map(|order| &order.symbol), + Some(&Symbol::new("BTCUSDT")) + ); + assert!(worker.pending_acks.contains_key(&order_id)); + assert_eq!( + worker + .tracked_orders + .get(&order_id) + .and_then(|order| order.venue), + Some(VenueId::BINANCE) + ); + assert_eq!( + worker + .tracked_orders + .get(&order_id) + .and_then(|order| order.account_id.as_ref()), + Some(&AccountId("binance-main".to_string())) + ); + assert_eq!(worker.stats.orders_placed, 2); + assert_eq!(worker.stats.orders_failed, 1); + let mut events = Vec::new(); + engine_queues.receive_events_into(&mut events); + assert_eq!( + events + .iter() + .filter(|event| matches!(event, ExecutionEvent::OrderNew { .. })) + .count(), + 1 + ); + + worker + .execution_streams + .push(Box::pin(futures::stream::iter( + [ + ExecutionEvent::Fill { + order_id: order_id.clone(), + price: Price::from_f64(100.0).unwrap(), + quantity: Quantity::from_f64(0.25).unwrap(), + timestamp: 1, + fill_id: "bitget-fill".to_string(), + }, + ExecutionEvent::OrderCanceled { + order_id: order_id.clone(), + timestamp: 2, + }, + ] + .into_iter() + .map(|event| IndexedExecutionStreamItem::Event { + client_idx: 1, + result: Ok::<_, HftError>(event), + }), + ))); + + assert_eq!(worker.poll_execution_events().await, 2); + assert!(worker.emergency_latched); + assert!(worker.stream_recovery_pending); + assert_eq!(worker.order_to_client.get(&order_id), Some(&0)); + assert_eq!( + worker + .tracked_orders + .get(&order_id) + .map(|order| order.remaining_quantity), + Some(Quantity::from_f64(1.0).unwrap()) + ); + assert!(worker.pending_acks.contains_key(&order_id)); + events.clear(); + engine_queues.receive_events_into(&mut events); + assert!(events.is_empty()); + } + #[tokio::test] async fn ack_timeout_latches_intake_and_keeps_the_order_pending_when_cancel_fails() { let state = Arc::new(StdMutex::new(MockExecutionState::default())); diff --git a/rust_hft/market-core/engine/src/lib.rs b/rust_hft/market-core/engine/src/lib.rs index 09a566121..e191aea11 100644 --- a/rust_hft/market-core/engine/src/lib.rs +++ b/rust_hft/market-core/engine/src/lib.rs @@ -17,8 +17,8 @@ use aggregation::{AggregationEngine, MarketView, TopNSnapshot}; use dataflow::{EventConsumer, IngestionConfig}; use execution_queues::EngineQueues; use hft_core::{ - now_micros, HftError, HftResult, LatencyStage, LatencyTracker, OrderType, Side, Symbol, - Timestamp, VenueId, VenueSymbol, + now_micros, AccountId, HftError, HftResult, LatencyStage, LatencyTracker, OrderType, Side, + Symbol, Timestamp, VenueId, VenueSymbol, }; use latency_monitor::{LatencyMonitor, LatencyMonitorConfig}; use ports::Trade as MarketTrade; @@ -57,6 +57,14 @@ pub enum TradingMode { Emergency, } +#[derive(Debug, Clone, Default)] +pub struct RuntimeTruthStatus { + pub reconciliation_complete: bool, + pub reconciliation_healthy: bool, + pub observed_at_us: u64, + pub account_id: Option, +} + /// 引擎運行統計（內部狀態） #[derive(Debug, Default)] pub struct EngineStats { @@ -72,6 +80,7 @@ pub struct EngineStats { pub market_events_dropped: u64, pub intents_dropped: u64, pub snapshot_publish_failed: u64, + pub data_integrity_gaps: u64, // Sentinel 控制 pub trading_mode: TradingMode, /// 引擎啟動時間 (微秒時間戳) @@ -231,6 +240,7 @@ pub struct Engine { market_snapshots: SnapshotContainer, /// 帳戶快照發佈容器 account_snapshots: SnapshotContainer, + runtime_truth_snapshots: SnapshotContainer, /// 註冊的策略 strategies: Vec>, /// 風控管理器（可選） @@ -321,6 +331,7 @@ impl Engine { aggregation_engine, market_snapshots: SnapshotContainer::new(initial_market_view), account_snapshots: SnapshotContainer::new(initial_account_view), + runtime_truth_snapshots: SnapshotContainer::new(RuntimeTruthStatus::default()), strategies: Vec::new(), risk_manager: None, venue_specs: VenueSpec::build_default_venue_specs(), @@ -341,6 +352,7 @@ impl Engine { market_events_dropped: 0, intents_dropped: 0, snapshot_publish_failed: 0, + data_integrity_gaps: 0, trading_mode: TradingMode::Normal, start_time_us: now_micros(), }, @@ -524,18 +536,55 @@ impl Engine { /// Compare the OMS source of truth with an authoritative exchange snapshot. pub fn reconcile_open_orders( &self, - exchange_orders: &[ports::OpenOrder], + snapshot: &execution_worker::WorkerReconcileSnapshot, ) -> ports::OrderReconciliationReport { - self.order_manager - .as_ref() - .map(|manager| manager.reconcile_with_exchange(exchange_orders)) - .unwrap_or_else(|| ports::OrderReconciliationReport { - exchange_only: exchange_orders + let Some(manager) = self.order_manager.as_ref() else { + return ports::OrderReconciliationReport { + exchange_only: snapshot + .clients .iter() + .filter_map(|client| client.open_orders.as_ref().ok()) + .flatten() .map(|order| order.order_id.clone()) .collect(), ..Default::default() - }) + }; + }; + + let local_orders = manager.export_state(); + let mut exchange_orders = Vec::new(); + let mut identity_conflicts = Vec::new(); + for client in &snapshot.clients { + let Some(open_orders) = client.open_orders.as_ref().ok() else { + continue; + }; + for exchange_order in open_orders { + let local_order = local_orders.get(&exchange_order.order_id).or_else(|| { + exchange_order + .client_order_id + .as_ref() + .and_then(|client_order_id| { + local_orders.values().find(|order| { + order.client_order_id.as_ref() == Some(client_order_id) + }) + }) + }); + let identity_matches = local_order.is_none_or(|order| { + order.symbol == exchange_order.symbol + && order.venue == client.venue + && self.order_account_map.get(&order.order_id) == client.account_id.as_ref() + }); + if identity_matches { + exchange_orders.push(exchange_order.clone()); + } else { + identity_conflicts.push(exchange_order.order_id.clone()); + } + } + } + + let mut report = manager.reconcile_with_exchange(&exchange_orders); + report.exchange_only.extend(identity_conflicts); + report } /// 取得指定策略的未結訂單 (order_id, symbol) 配對，供外部控制/撤單使用 @@ -551,6 +600,15 @@ impl Engine { /// 導入 OMS 狀態（供恢復/持久化使用） pub fn import_oms_state(&mut self, state: HashMap) { + self.order_account_map.clear(); + self.order_account_map + .extend(state.iter().filter_map(|(order_id, record)| { + record + .strategy_id + .as_ref() + .and_then(|strategy_id| self.strategy_account_mapping.get(strategy_id)) + .map(|account_id| (order_id.clone(), account_id.clone())) + })); if let Some(om) = &mut self.order_manager { om.import_state(state); } @@ -622,6 +680,7 @@ impl Engine { } // 同步引擎統計（以 Gauge） let s = self.get_statistics(); + let runtime_truth = self.runtime_truth_snapshots.load(); let export = infra_metrics::EngineStatisticsExport { cycle_count: s.cycle_count, execution_events_processed: s.execution_events_processed, @@ -630,6 +689,14 @@ impl Engine { orders_filled: s.orders_filled, orders_rejected: s.orders_rejected, orders_canceled: s.orders_canceled, + runtime_truth_observed_at_us: runtime_truth.observed_at_us, + reconciliation_complete: runtime_truth.reconciliation_complete, + reconciliation_healthy: runtime_truth.reconciliation_healthy, + risk_halted: matches!( + self.stats.trading_mode, + TradingMode::Paused | TradingMode::Emergency + ), + data_integrity_gaps: s.data_integrity_gaps, }; infra_metrics::MetricsRegistry::global().update_engine_statistics(&export); } @@ -815,6 +882,14 @@ impl Engine { self.account_snapshots.reader() } + pub fn runtime_truth_reader(&self) -> Arc> { + self.runtime_truth_snapshots.reader() + } + + pub(crate) fn publish_runtime_truth_status(&mut self, status: RuntimeTruthStatus) { + self.runtime_truth_snapshots.store(Arc::new(status)); + } + /// 提交訂單意圖到執行隊列（用於 dry-run 測試） pub fn submit_order_intent(&mut self, intent: ports::OrderIntent) -> Result<(), HftError> { self.ensure_accepting_new_intents()?; @@ -979,6 +1054,10 @@ impl Engine { } for event in aggregation_events.drain(..) { + if matches!(event, ports::MarketEvent::Disconnect { .. }) { + self.stats.data_integrity_gaps = + self.stats.data_integrity_gaps.saturating_add(1); + } let book = Self::event_venue_symbol(&event).and_then(|key| { self.aggregation_engine .orderbooks @@ -1772,20 +1851,36 @@ impl Engine { .unwrap_or((0, 0)); // 從 Portfolio 獲取 PnL 和回撤（如果有） - let (pnl, unrealized_pnl, drawdown_pct, max_drawdown_pct, high_water_mark) = - if let Some(pm) = &self.portfolio_manager { - let av = pm.reader().load(); - let total_pnl = av.total_pnl(); - ( - total_pnl.to_string().parse::().unwrap_or(0.0), - av.unrealized_pnl.to_string().parse::().unwrap_or(0.0), - av.drawdown_pct, - av.max_drawdown_pct, - av.high_water_mark.to_string().parse::().unwrap_or(0.0), - ) - } else { - (0.0, 0.0, 0.0, 0.0, 0.0) - }; + let ( + pnl, + unrealized_pnl, + drawdown_pct, + max_drawdown_pct, + high_water_mark, + position_count, + notional_value, + ) = if let Some(pm) = &self.portfolio_manager { + let av = pm.reader().load(); + let total_pnl = av.total_pnl(); + let notional = av + .positions + .values() + .map(|position| { + (position.avg_price.0 * position.quantity.0 + position.unrealized_pnl).abs() + }) + .sum::(); + ( + total_pnl.to_string().parse::().unwrap_or(0.0), + av.unrealized_pnl.to_string().parse::().unwrap_or(0.0), + av.drawdown_pct, + av.max_drawdown_pct, + av.high_water_mark.to_string().parse::().unwrap_or(0.0), + av.positions.len() as i64, + notional.to_string().parse::().unwrap_or(0.0), + ) + } else { + (0.0, 0.0, 0.0, 0.0, 0.0, 0, 0.0) + }; SentinelStats { latency_p99_us, @@ -1795,6 +1890,8 @@ impl Engine { drawdown_pct, max_drawdown_pct, high_water_mark, + position_count, + notional_value, } } @@ -1901,6 +1998,10 @@ impl Engine { orders_filled: self.stats.orders_filled, orders_rejected: self.stats.orders_rejected, orders_canceled: self.stats.orders_canceled, + market_events_dropped: self.stats.market_events_dropped, + intents_dropped: self.stats.intents_dropped, + snapshot_publish_failed: self.stats.snapshot_publish_failed, + data_integrity_gaps: self.stats.data_integrity_gaps, latency, uptime_seconds, } @@ -2058,6 +2159,10 @@ pub struct SentinelStats { pub max_drawdown_pct: f64, /// 高水位標記 pub high_water_mark: f64, + /// Number of authoritative open positions. + pub position_count: i64, + /// Gross marked notional across authoritative positions. + pub notional_value: f64, } /// 引擎延遲統計信息 @@ -2097,6 +2202,10 @@ pub struct EngineStatistics { pub orders_filled: u64, pub orders_rejected: u64, pub orders_canceled: u64, + pub market_events_dropped: u64, + pub intents_dropped: u64, + pub snapshot_publish_failed: u64, + pub data_integrity_gaps: u64, /// 延遲統計 pub latency: EngineLatencyStats, /// 引擎運行時間（秒） diff --git a/rust_hft/market-core/ports/src/traits.rs b/rust_hft/market-core/ports/src/traits.rs index d070703d8..09fc0df8d 100644 --- a/rust_hft/market-core/ports/src/traits.rs +++ b/rust_hft/market-core/ports/src/traits.rs @@ -242,6 +242,9 @@ pub struct Position { pub quantity: Quantity, pub avg_price: Price, pub unrealized_pnl: rust_decimal::Decimal, + /// Realized PnL accumulated while the current position lifecycle is open. + #[serde(default)] + pub realized_pnl: rust_decimal::Decimal, } /// 策略接口 diff --git a/rust_hft/market-core/runtime/src/ipc_handler.rs b/rust_hft/market-core/runtime/src/ipc_handler.rs index 4fa5e455e..14765b08d 100644 --- a/rust_hft/market-core/runtime/src/ipc_handler.rs +++ b/rust_hft/market-core/runtime/src/ipc_handler.rs @@ -235,9 +235,7 @@ impl CommandHandler for SystemCommandHandler { average_price: pos.avg_price.0, market_value: pos.avg_price.0 * pos.quantity.0 + pos.unrealized_pnl, unrealized_pnl: pos.unrealized_pnl, - realized_pnl: rust_decimal::Decimal::ZERO, // TODO: 逐部位已实现损益追踪需要扩展 ports::Position 结构 - // 当前系统仅在 AccountView 层面追踪总已实现损益 - // 未来改进：在 Position 中添加 realized_pnl 字段，并在 PortfolioCore 中追踪每个仓位的平仓损益 + realized_pnl: pos.realized_pnl, last_update: std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) .unwrap_or_default() @@ -1072,6 +1070,7 @@ mod tests { quantity: Quantity(Decimal::from(2)), avg_price: Price(Decimal::new(51, 2)), unrealized_pnl: Decimal::new(5, 2), + realized_pnl: Decimal::ZERO, }])), recent_fills: Some(Ok(vec![ports::AccountFill { fill_id: "fill-1".to_string(), diff --git a/rust_hft/risk-control/portfolio-core/src/lib.rs b/rust_hft/risk-control/portfolio-core/src/lib.rs index 8ef92ae8e..b545f6642 100644 --- a/rust_hft/risk-control/portfolio-core/src/lib.rs +++ b/rust_hft/risk-control/portfolio-core/src/lib.rs @@ -244,6 +244,7 @@ impl Portfolio { quantity: Quantity::zero(), avg_price: Price::zero(), unrealized_pnl: Decimal::ZERO, + realized_pnl: Decimal::ZERO, }); let old_qty = pos.quantity.0; @@ -267,7 +268,9 @@ impl Portfolio { } else { -Decimal::ONE }; - self.view.realized_pnl += (price.0 - pos.avg_price.0) * closed_quantity * direction; + let realized_delta = (price.0 - pos.avg_price.0) * closed_quantity * direction; + pos.realized_pnl += realized_delta; + self.view.realized_pnl += realized_delta; if new_qty == Decimal::ZERO { pos.avg_price = Price::zero(); @@ -454,6 +457,7 @@ mod tests { let position = view.positions.get(&symbol).unwrap(); assert_eq!(position.quantity.0, Decimal::from(2)); assert_eq!(position.avg_price.0, Decimal::from(110)); + assert_eq!(position.realized_pnl, Decimal::from(30)); assert_eq!(view.realized_pnl, Decimal::from(30)); fill(&mut portfolio, "S-2", "f4", &symbol, Side::Sell, 90, 2); @@ -485,6 +489,7 @@ mod tests { let position = view.positions.get(&symbol).unwrap(); assert_eq!(position.quantity.0, Decimal::from(-1)); assert_eq!(position.avg_price.0, Decimal::from(110)); + assert_eq!(position.realized_pnl, Decimal::from(20)); assert_eq!(view.realized_pnl, Decimal::from(20)); fill(&mut portfolio, "B-2", "f4", &symbol, Side::Buy, 100, 2); @@ -493,6 +498,7 @@ mod tests { let position = view.positions.get(&symbol).unwrap(); assert_eq!(position.quantity.0, Decimal::ONE); assert_eq!(position.avg_price.0, Decimal::from(100)); + assert_eq!(position.realized_pnl, Decimal::from(30)); assert_eq!(view.realized_pnl, Decimal::from(30)); assert_eq!(view.cash_balance, Decimal::from(930)); assert_eq!(view.equity(), Decimal::from(1030)); From d44f218dda8e03c1d8f28f14a15905d0f7111e23 Mon Sep 17 00:00:00 2001 From: proerror Date: Thu, 16 Jul 2026 14:01:05 +0800 Subject: [PATCH 2/5] fix(metrics): align account truth export schema --- rust_hft/infra-services/core/metrics/src/lib.rs | 7 +++++++ .../market-core/runtime/src/system_builder.rs | 17 ++--------------- 2 files changed, 9 insertions(+), 15 deletions(-) diff --git a/rust_hft/infra-services/core/metrics/src/lib.rs b/rust_hft/infra-services/core/metrics/src/lib.rs index 4593182f0..49256f1b9 100644 --- a/rust_hft/infra-services/core/metrics/src/lib.rs +++ b/rust_hft/infra-services/core/metrics/src/lib.rs @@ -86,6 +86,13 @@ pub struct EngineStatisticsExport { pub orders_filled: u64, pub orders_rejected: u64, pub orders_canceled: u64, + // Account-truth facts are emitted by the engine. The readiness policy that + // interprets them remains in the runtime-health layer. + pub runtime_truth_observed_at_us: u64, + pub reconciliation_complete: bool, + pub reconciliation_healthy: bool, + pub risk_halted: bool, + pub data_integrity_gaps: u64, } impl MetricsRegistry { diff --git a/rust_hft/market-core/runtime/src/system_builder.rs b/rust_hft/market-core/runtime/src/system_builder.rs index 14d337e52..011aec702 100644 --- a/rust_hft/market-core/runtime/src/system_builder.rs +++ b/rust_hft/market-core/runtime/src/system_builder.rs @@ -1320,6 +1320,8 @@ impl SystemRuntime { interval.tick().await; let (cash, pos, unr, rlz, stats) = { let eng = engine_arc.lock().await; + #[cfg(feature = "metrics")] + eng.sync_latency_metrics_to_prometheus(); let av = eng.get_account_view(); let st = eng.get_statistics(); ( @@ -1330,21 +1332,6 @@ impl SystemRuntime { st, ) }; - // 導出引擎統計到 Prometheus（僅在 metrics feature 啟用時） - #[cfg(feature = "metrics")] - { - infra_metrics::MetricsRegistry::global().update_engine_statistics( - &infra_metrics::EngineStatisticsExport { - cycle_count: stats.cycle_count, - execution_events_processed: stats.execution_events_processed, - orders_submitted: stats.orders_submitted, - orders_ack: stats.orders_ack, - orders_filled: stats.orders_filled, - orders_rejected: stats.orders_rejected, - orders_canceled: stats.orders_canceled, - }, - ); - } // 當引擎停止時，狀態任務退出，避免 Ctrl-C 卡住 if !stats.is_running { break; From a0994d72860182f271689b282f73842976f72bb1 Mon Sep 17 00:00:00 2001 From: proerror Date: Thu, 16 Jul 2026 13:30:45 +0800 Subject: [PATCH 3/5] fix(risk): enforce batch exposure and tokenized policy --- .../adapters/adapter-ondo-perps/src/lib.rs | 1 + rust_hft/market-core/core/src/types.rs | 18 + .../tests/tokenized_security_execution_e2e.rs | 9 +- rust_hft/market-core/ports/src/traits.rs | 27 +- .../runtime/src/exposure_projection.rs | 181 +++++++++ rust_hft/market-core/runtime/src/lib.rs | 1 + .../runtime/src/portfolio_manager.rs | 207 ++++++++++ .../runtime/src/risk_manager_factory.rs | 361 +++++++++++++++++- .../market-core/runtime/src/system_builder.rs | 9 + .../risk/src/default_risk_manager.rs | 63 ++- rust_hft/risk-control/risk/src/sentinel.rs | 61 ++- 11 files changed, 904 insertions(+), 34 deletions(-) create mode 100644 rust_hft/market-core/runtime/src/exposure_projection.rs 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 011aec702..f0c8607ef 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!( From d6e32c25f1bcfe17ddcbbce7c9baf9648485a720 Mon Sep 17 00:00:00 2001 From: proerror Date: Thu, 16 Jul 2026 15:34:17 +0800 Subject: [PATCH 4/5] ci: rerun checks after main retarget From 24771fd6dd9a23abc2fab2916cea55acb4eb542b Mon Sep 17 00:00:00 2001 From: proerror Date: Thu, 16 Jul 2026 15:43:07 +0800 Subject: [PATCH 5/5] fix(risk): align sentinel stats schema --- rust_hft/apps/live/src/helpers/sentinel.rs | 1 - 1 file changed, 1 deletion(-) diff --git a/rust_hft/apps/live/src/helpers/sentinel.rs b/rust_hft/apps/live/src/helpers/sentinel.rs index 924e34918..91152606c 100644 --- a/rust_hft/apps/live/src/helpers/sentinel.rs +++ b/rust_hft/apps/live/src/helpers/sentinel.rs @@ -104,7 +104,6 @@ async fn run_sentinel_loop( position_count: active_orders as i64, notional_value: 0.0, order_rate: 0.0, - ws_reconnect_count: 0, data_gap_count: 0, };