分离过程事件分发与结果保留

This commit is contained in:
boris
2026-08-27 09:04:48 +08:00
parent 00ec7a6d55
commit 9db2a9f79c
4 changed files with 124 additions and 6 deletions
+43 -3
View File
@@ -71,6 +71,23 @@ impl Default for FuturesValidationConfig {
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ProcessEventRetention {
/// Retain every phase and business event in the returned result.
All,
/// Retain only lifecycle events useful for a durable business audit.
Business,
/// Dispatch events to listeners and the strategy, but do not retain them
/// in the returned result.
None,
}
impl Default for ProcessEventRetention {
fn default() -> Self {
Self::All
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DailyEquityPoint {
#[serde(with = "date_format")]
@@ -425,6 +442,7 @@ pub struct BacktestEngine<S, C, R> {
cash_dividends_enabled: bool,
cash_dividend_adjusts_cost_basis: bool,
process_event_bus: ProcessEventBus,
process_event_retention: ProcessEventRetention,
dynamic_universe: Option<BTreeSet<String>>,
subscriptions: BTreeSet<String>,
futures_account: Option<FuturesAccountState>,
@@ -455,6 +473,7 @@ impl<S, C, R> BacktestEngine<S, C, R> {
cash_dividends_enabled: true,
cash_dividend_adjusts_cost_basis: true,
process_event_bus: ProcessEventBus::new(),
process_event_retention: ProcessEventRetention::All,
dynamic_universe: None,
subscriptions: BTreeSet::new(),
futures_account: None,
@@ -488,6 +507,11 @@ impl<S, C, R> BacktestEngine<S, C, R> {
self
}
pub fn with_process_event_retention(mut self, retention: ProcessEventRetention) -> Self {
self.process_event_retention = retention;
self
}
pub fn with_cash_dividends(mut self, enabled: bool) -> Self {
self.cash_dividends_enabled = enabled;
self
@@ -2118,7 +2142,7 @@ where
let holding_count = holdings_for_day.len();
result.daily_holdings.extend(holdings_for_day);
let progress_process_start = result.process_events.len();
result.process_events.append(&mut process_events);
self.retain_process_events(&mut result.process_events, &mut process_events);
let aggregate_cash = self.aggregate_cash(&portfolio);
let aggregate_market_value = self.aggregate_market_value(&portfolio);
let aggregate_total_equity = self.aggregate_total_equity(&portfolio);
@@ -3193,7 +3217,7 @@ where
let holding_count = holdings_for_day.len();
result.daily_holdings.extend(holdings_for_day);
let progress_process_start = result.process_events.len();
result.process_events.append(&mut process_events);
self.retain_process_events(&mut result.process_events, &mut process_events);
let aggregate_cash = self.aggregate_cash(&portfolio);
let aggregate_market_value = self.aggregate_market_value(&portfolio);
let aggregate_total_equity = self.aggregate_total_equity(&portfolio);
@@ -3280,7 +3304,23 @@ where
result.fills.append(&mut report.fill_events);
result.position_events.append(&mut report.position_events);
result.account_events.append(&mut report.account_events);
result.process_events.append(&mut report.process_events);
self.retain_process_events(&mut result.process_events, &mut report.process_events);
}
fn retain_process_events(
&self,
target: &mut Vec<ProcessEvent>,
incoming: &mut Vec<ProcessEvent>,
) {
match self.process_event_retention {
ProcessEventRetention::All => target.append(incoming),
ProcessEventRetention::Business => target.extend(
incoming
.drain(..)
.filter(|event| event.kind.is_business_lifecycle()),
),
ProcessEventRetention::None => incoming.clear(),
}
}
fn apply_corporate_actions(
+41 -1
View File
@@ -364,6 +364,38 @@ impl ProcessEventKind {
Self::AccountManagementFee => "account_management_fee",
}
}
/// Returns whether the event is part of the durable business lifecycle
/// audit. Phase boundary events are useful during interactive debugging,
/// but retaining every minute phase marker for a long run is unnecessary.
pub fn is_business_lifecycle(&self) -> bool {
matches!(
*self,
Self::PreScheduled
| Self::PostScheduled
| Self::PreOnDay
| Self::OnDay
| Self::PostOnDay
| Self::OrderPendingNew
| Self::OrderCreationPass
| Self::OrderCreationReject
| Self::OrderPendingCancel
| Self::OrderCancellationPass
| Self::OrderCancellationReject
| Self::OrderPendingUpdate
| Self::OrderUpdatePass
| Self::OrderUpdateReject
| Self::OrderUnsolicitedUpdate
| Self::Trade
| Self::UniverseUpdated
| Self::UniverseSubscribed
| Self::UniverseUnsubscribed
| Self::AccountDepositWithdraw
| Self::AccountFinanceRepay
| Self::AccountManagementFee
| Self::Settlement
)
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
@@ -384,7 +416,7 @@ pub struct ProcessEvent {
mod tests {
use chrono::{NaiveDate, NaiveDateTime};
use super::{FillEvent, OrderEvent, OrderSide, OrderStatus};
use super::{FillEvent, OrderEvent, OrderSide, OrderStatus, ProcessEventKind};
fn order_event(status: OrderStatus, filled_quantity: u32) -> OrderEvent {
OrderEvent {
@@ -484,4 +516,12 @@ mod tests {
"2025-01-02 10:18:03"
);
}
#[test]
fn process_event_business_lifecycle_filter_keeps_audit_events_only() {
assert!(ProcessEventKind::OrderUpdateReject.is_business_lifecycle());
assert!(ProcessEventKind::Settlement.is_business_lifecycle());
assert!(!ProcessEventKind::PreMinute.is_business_lifecycle());
assert!(!ProcessEventKind::PostBar.is_business_lifecycle());
}
}
+1
View File
@@ -39,6 +39,7 @@ pub use engine::{
AnalyzerTradeRow, BacktestConfig, BacktestDayProgress, BacktestEngine, BacktestError,
BacktestResult, BacktestTerminalAssetClass, BacktestTerminalAudit, BacktestTerminalOpenOrder,
BacktestTerminalStatus, DailyEquityPoint, ExecutionQuoteRequest, FuturesValidationConfig,
ProcessEventRetention,
};
pub use event_bus::{BacktestProcessMod, BacktestProcessModLoader, ProcessEventBus};
pub use events::{