From a39210b4bdec35258d1f6facd85fb58abc71594c Mon Sep 17 00:00:00 2001 From: proerror Date: Thu, 16 Jul 2026 13:29:17 +0800 Subject: [PATCH 1/2] 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/2] 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;