合并组合亏损控制器与因子内核依赖

This commit is contained in:
boris
2026-09-09 10:40:26 +08:00
8 changed files with 854 additions and 9 deletions
+1
View File
@@ -13,5 +13,6 @@ rayon.workspace = true
rhai.workspace = true
serde.workspace = true
serde_json.workspace = true
sha2.workspace = true
thiserror.workspace = true
ta-lib = { git = "https://github.com/TA-Lib/ta-lib.git", rev = "dd5a90259a3f9e04e2da9f38bf0719a841b40108" }
+2
View File
@@ -17,6 +17,7 @@ pub mod platform_expr_strategy;
pub mod platform_runtime_schema;
pub mod platform_strategy_spec;
pub mod portfolio;
pub mod portfolio_loss;
pub mod risk_control;
pub mod rules;
pub mod scheduler;
@@ -86,6 +87,7 @@ pub use platform_strategy_spec::{
platform_expr_config_from_value, validate_strategy_risk_policy_fields,
};
pub use portfolio::{CashReceivable, HoldingSummary, PendingCashFlow, PortfolioState, Position};
pub use portfolio_loss::{ClosedPortfolioSession, PortfolioLossConfig, PortfolioLossDecision, PortfolioLossError, PortfolioLossState};
pub use risk_control::{
ChinaAShareRiskControl, FidcRiskControlConfig, FidcRiskDecisionAudit, RiskCheckScope,
StaticRiskRuleConfig, TradingConstraintConfig,
+195 -8
View File
@@ -3,7 +3,8 @@ use std::collections::{BTreeMap, BTreeSet};
use std::sync::Arc;
use ahash::{AHashMap, AHashSet};
use chrono::{Datelike, Duration, NaiveDate, NaiveDateTime, NaiveTime, Timelike};
use chrono::{Datelike, Duration, FixedOffset, NaiveDate, NaiveDateTime, NaiveTime, TimeZone, Timelike, Utc};
use sha2::{Digest, Sha256};
use rhai::{AST, Dynamic, Engine, ImmutableString, Map, Scope};
use crate::broker::{MatchingType, RebalanceCashMode, SlippageModel};
@@ -16,7 +17,7 @@ use crate::data::{
decision_market_cap_bn,
};
use crate::engine::BacktestError;
use crate::events::OrderSide;
use crate::events::{OrderSide, ProcessEvent, ProcessEventKind};
use crate::fixed_point::FixedMoney;
use crate::futures::{
FuturesContractSpec, FuturesDirection, FuturesOrderIntent, FuturesPositionEffect,
@@ -26,6 +27,7 @@ use crate::numeric_expr_vm::{
Scratch as NumericVmScratch, Value as NumericVmValue, ValueType as NumericVmValueType,
};
use crate::portfolio::PortfolioState;
use crate::portfolio_loss::{ClosedPortfolioSession, PortfolioLossConfig, PortfolioLossState};
use crate::risk_control::{ChinaAShareRiskControl, FidcRiskControlConfig, FidcRiskDecisionAudit};
use crate::scheduler::{ScheduleRule, ScheduleStage, ScheduleTimeRule, Scheduler};
use crate::strategy::{
@@ -179,6 +181,22 @@ impl PlatformPortfolioDrawdownController {
}
}
fn portfolio_gross_exposure(portfolio: &PortfolioState) -> Result<f64, BacktestError> {
let equity = portfolio.total_equity();
let market_value: f64 = portfolio.positions().values().map(|position| position.market_value().abs()).sum();
if !equity.is_finite() || equity <= 0.0 || !market_value.is_finite() {
return Err(BacktestError::Execution("portfolio loss requires finite positive accounting equity".to_owned()));
}
Ok(market_value / equity)
}
fn portfolio_loss_decision_at(ctx: &StrategyContext<'_>) -> chrono::DateTime<Utc> {
let local = ctx.active_datetime.filter(|value| value.date() == ctx.execution_date)
.unwrap_or_else(|| ctx.execution_date.and_hms_opt(9, 30, 0).unwrap());
FixedOffset::east_opt(8 * 3600).unwrap().from_local_datetime(&local)
.single().unwrap().with_timezone(&Utc)
}
fn validated_target_scales(scales: &[(String, f64)]) -> Result<Vec<(String, f64)>, BacktestError> {
if scales.is_empty() {
return Ok(Vec::new());
@@ -610,6 +628,7 @@ pub struct PlatformExprStrategyConfig {
pub exposure_expr: String,
pub position_exposure_schedule: BTreeMap<NaiveDate, f64>,
pub portfolio_drawdown_control: Option<PlatformPortfolioDrawdownControlConfig>,
pub portfolio_loss_control: Option<PortfolioLossConfig>,
pub stop_loss_expr: String,
pub take_profit_expr: String,
pub position_target_rules: Vec<PlatformPositionTargetRule>,
@@ -690,6 +709,7 @@ impl PlatformExprStrategyConfig {
exposure_expr: "1.0".to_string(),
position_exposure_schedule: BTreeMap::new(),
portfolio_drawdown_control: None,
portfolio_loss_control: None,
stop_loss_expr: String::new(),
take_profit_expr: String::new(),
position_target_rules: Vec::new(),
@@ -1335,6 +1355,8 @@ pub struct PlatformExprStrategy {
last_target_order: Option<Vec<String>>,
last_trading_ratio: Option<f64>,
portfolio_drawdown_controller: Option<PlatformPortfolioDrawdownController>,
portfolio_loss_state: Option<PortfolioLossState>,
portfolio_loss_opening: Option<(NaiveDate, f64, f64)>,
pending_highlimit_holdings: BTreeSet<String>,
pending_full_close_symbols: BTreeSet<String>,
position_entry_dates: BTreeMap<String, NaiveDate>,
@@ -1433,6 +1455,21 @@ fn completed_session_factor_date(
}
impl PlatformExprStrategy {
pub fn portfolio_loss_state(&self) -> Option<&PortfolioLossState> {
self.portfolio_loss_state.as_ref()
}
pub fn restore_portfolio_loss_state(&mut self, state: PortfolioLossState) -> Result<(), BacktestError> {
let expected = self.config.portfolio_loss_control.as_ref().ok_or_else(|| BacktestError::Execution(
"portfolio loss state supplied for a strategy without that control".to_owned()))?;
state.validate(expected).map_err(|error| BacktestError::Execution(error.to_string()))?;
if self.portfolio_loss_opening.is_some() {
return Err(BacktestError::Execution("cannot restore portfolio loss state during an open session".to_owned()));
}
self.portfolio_loss_state = Some(state);
Ok(())
}
fn market_cap_storage_to_strategy_unit(value: f64) -> f64 {
value
}
@@ -1729,6 +1766,8 @@ impl PlatformExprStrategy {
last_target_order: None,
last_trading_ratio: None,
portfolio_drawdown_controller,
portfolio_loss_state: None,
portfolio_loss_opening: None,
pending_highlimit_holdings: BTreeSet::new(),
pending_full_close_symbols: BTreeSet::new(),
position_entry_dates: BTreeMap::new(),
@@ -8485,12 +8524,18 @@ impl PlatformExprStrategy {
)
.unwrap_or(strategy_exposure)
.clamp(0.0, 1.0);
let Some(controller) = self.portfolio_drawdown_controller.as_mut() else {
return Ok(risk_on_exposure);
};
controller
.update(ctx.decision_date, day.total_value, risk_on_exposure)
.map(|decision| decision.target_exposure.clamp(0.0, 1.0))
let mut exposure = risk_on_exposure;
if let Some(controller) = self.portfolio_drawdown_controller.as_mut() {
exposure = controller.update(ctx.decision_date, day.total_value, exposure)?.target_exposure;
}
if self.config.portfolio_loss_control.is_some() {
let state = self.portfolio_loss_state.as_mut().ok_or_else(|| BacktestError::Execution(
"portfolio loss state must be initialized or restored before planning".to_owned()))?;
let previous = ctx.data.previous_trading_date(ctx.execution_date, 1);
exposure = state.decide(ctx.execution_date, previous, portfolio_loss_decision_at(ctx), exposure)
.map_err(|error| BacktestError::Execution(error.to_string()))?.target_exposure;
}
Ok(exposure.clamp(0.0, 1.0))
}
fn market_cap_band(
@@ -12144,6 +12189,59 @@ impl Strategy for PlatformExprStrategy {
self.config.strategy_name.as_str()
}
fn before_trading(&mut self, ctx: &StrategyContext<'_>) -> Result<(), BacktestError> {
let Some(config) = self.config.portfolio_loss_control.clone() else { return Ok(()); };
if ctx.futures_account.is_some() {
return Err(BacktestError::Execution("portfolio loss control currently requires equity-only accounting".to_owned()));
}
if self.portfolio_loss_state.is_none() {
self.portfolio_loss_state = Some(PortfolioLossState::new(config, ctx.execution_date)
.map_err(|error| BacktestError::Execution(error.to_string()))?);
}
if let Some((date, _, _)) = self.portfolio_loss_opening {
if date == ctx.execution_date { return Ok(()); }
return Err(BacktestError::Execution("portfolio loss previous session was not finalized".to_owned()));
}
let state = self.portfolio_loss_state.as_ref().unwrap();
let start_nav = state.last_session().map(|row| row.end_unit_nav).unwrap_or_else(|| ctx.portfolio.unit_net_value());
let gross = portfolio_gross_exposure(ctx.portfolio)?;
self.portfolio_loss_opening = Some((ctx.execution_date, start_nav, gross));
// Advance the daily risk clock even when the selection schedule is not
// due. The actual current exposure budget is applied during planning.
self.portfolio_loss_state.as_mut().unwrap().decide(ctx.execution_date,
ctx.data.previous_trading_date(ctx.execution_date, 1), portfolio_loss_decision_at(ctx), 1.0)
.map_err(|error| BacktestError::Execution(error.to_string()))?;
Ok(())
}
fn on_process_event(&mut self, ctx: &StrategyContext<'_>, event: &ProcessEvent) -> Result<(), BacktestError> {
if self.config.portfolio_loss_control.is_none() || event.kind != ProcessEventKind::PostSettlement { return Ok(()); }
let Some((date, start_nav, start_gross)) = self.portfolio_loss_opening else {
return Err(BacktestError::Execution("portfolio loss settlement has no opening accounting snapshot".to_owned()));
};
if date != ctx.execution_date {
return Err(BacktestError::Execution("portfolio loss settlement date differs from opening snapshot".to_owned()));
}
let end_nav = ctx.portfolio.unit_net_value();
let end_gross = portfolio_gross_exposure(ctx.portfolio)?;
let state = self.portfolio_loss_state.as_mut().unwrap();
let previous = state.last_session().map(|row| row.date);
let mut hash = Sha256::new();
hash.update(b"fidc.engine-finalized-portfolio-session/v1\0");
hash.update(self.config.strategy_name.as_bytes());
hash.update(date.to_string().as_bytes());
for value in [start_nav, end_nav, start_gross, end_gross] { hash.update(value.to_bits().to_le_bytes()); }
state.observe(ClosedPortfolioSession {
date, previous_session_date: previous,
available_at: date.and_hms_opt(7, 30, 0).unwrap().and_utc(),
start_unit_nav: start_nav, end_unit_nav: end_nav,
start_gross_exposure: start_gross, end_gross_exposure: end_gross,
source_sha256: format!("{:x}", hash.finalize()),
}).map_err(|error| BacktestError::Execution(error.to_string()))?;
self.portfolio_loss_opening = None;
Ok(())
}
fn initial_subscriptions(&self) -> BTreeSet<String> {
self.config.initial_subscriptions.clone()
}
@@ -14006,6 +14104,10 @@ impl PlatformExprStrategy {
{
diagnostics.push(diagnostic);
}
if let Some(decision) = self.portfolio_loss_state.as_ref().and_then(PortfolioLossState::last_decision) {
diagnostics.push(format!("portfolio_loss_control {}", serde_json::to_string(decision)
.map_err(|error| BacktestError::Execution(error.to_string()))?));
}
let notes = vec![
format!("stock_list={}", stock_list.len()),
@@ -14535,6 +14637,91 @@ mod tests {
.expect("single-symbol platform dataset")
}
#[test]
fn portfolio_loss_observes_finalized_nav_after_fees_and_cash_flows() {
use std::sync::Mutex;
use chrono::Duration;
use crate::{BacktestConfig, BacktestEngine, BacktestError, BrokerSimulator,
ChinaAShareCostModel, ChinaEquityRuleHooks, ClosedPortfolioSession,
PortfolioLossConfig, PriceField, StrategyDecision};
struct Capture {
inner: PlatformExprStrategy,
first: NaiveDate,
rows: Arc<Mutex<Vec<(ClosedPortfolioSession, crate::portfolio_loss::PortfolioLossDecision)>>>,
}
impl Strategy for Capture {
fn name(&self) -> &str { "portfolio-loss-lifecycle-test" }
fn requires_minute_callbacks(&self) -> bool { false }
fn before_trading(&mut self, ctx: &StrategyContext<'_>) -> Result<(), BacktestError> {
self.inner.before_trading(ctx)
}
fn on_day(&mut self, ctx: &StrategyContext<'_>) -> Result<StrategyDecision, BacktestError> {
let mut decision = self.inner.on_day(ctx)?;
if ctx.execution_date == self.first {
decision.order_intents.push(OrderIntent::SetManagementFeeRate { rate: 0.001, reason: "fee accounting test".to_owned() });
}
if ctx.execution_date == self.first + Duration::days(5) {
decision.order_intents.push(OrderIntent::DepositWithdraw { amount: 10_000.0, receiving_days: 0, reason: "unit NAV flow test".to_owned() });
}
Ok(decision)
}
fn on_process_event(&mut self, ctx: &StrategyContext<'_>, event: &ProcessEvent) -> Result<(), BacktestError> {
self.inner.on_process_event(ctx, event)?;
if event.kind == ProcessEventKind::PostSettlement {
let state = self.inner.portfolio_loss_state().unwrap();
self.rows.lock().unwrap().push((state.last_session().unwrap().clone(), state.last_decision().unwrap().clone()));
}
Ok(())
}
}
let first = d(2023, 1, 3);
let dates = (0..25).map(|day| first + Duration::days(day)).collect::<Vec<_>>();
let mut parts = single_symbol_platform_data(&dates, "000001.SZ").snapshot_components();
for (index, row) in parts.market.iter_mut().enumerate() {
let price = (1000.0 * 0.99_f64.powi(index as i32)).round() / 100.0;
row.day_open = price; row.open = price; row.high = price; row.low = price;
row.close = price; row.last_price = price; row.bid1 = price; row.ask1 = price;
row.prev_close = price / 0.99; row.upper_limit = price * 1.1; row.lower_limit = price * 0.9;
}
let data = DataSet::from_components(parts.instruments, parts.market, parts.factors, parts.candidates, parts.benchmarks).unwrap();
let mut config = PlatformExprStrategyConfig::generic();
config.universe_include = Some(BTreeSet::from(["000001.SZ".to_owned()]));
config.signal_symbol = "000001.SZ".to_owned();
config.benchmark_symbol = "000852.SH".to_owned();
config.stock_filter_expr = "true".to_owned(); config.rank_expr = "1.0".to_owned();
config.selection_limit_expr = "1".to_owned(); config.max_positions = 1;
config.market_cap_lower_expr = "0.0".to_owned(); config.market_cap_upper_expr = "1000.0".to_owned();
config.exposure_expr = "0.9".to_owned(); config.refresh_rate = 1; config.refresh_rate_expr = "1".to_owned();
config.rebalance_existing_positions = true;
config.portfolio_loss_control = Some(PortfolioLossConfig { lookback: 10, loss_trigger: 0.05, floor_exposure: 0.2, cooldown_trading_days: 3 });
let rows = Arc::new(Mutex::new(Vec::new()));
let strategy = Capture { inner: PlatformExprStrategy::new(config), first, rows: Arc::clone(&rows) };
let broker = BrokerSimulator::new(ChinaAShareCostModel::default(), ChinaEquityRuleHooks)
.with_matching_type(MatchingType::CurrentBarClose);
let mut engine = BacktestEngine::new(data, strategy, broker, BacktestConfig {
initial_cash: 10_000.0, benchmark_code: "000852.SH".to_owned(), start_date: Some(first),
end_date: dates.last().copied(), decision_lag_trading_days: 0, execution_price_field: PriceField::Close,
});
let result = engine.run().unwrap();
let records = rows.lock().unwrap();
assert_eq!(records.len(), dates.len());
assert!(records.iter().any(|(_, decision)| decision.newly_triggered), "risk sessions={:?}",
records.iter().map(|(session, decision)| (session.date, session.start_unit_nav, session.end_unit_nav,
session.start_gross_exposure, session.end_gross_exposure, decision.trailing_unit_return, decision.target_exposure)).collect::<Vec<_>>());
assert!(result.fills.len() > 1, "fills={} orders={:?} diagnostics={:?}", result.fills.len(),
result.order_events.iter().take(4).collect::<Vec<_>>(),
result.equity_curve.iter().take(3).map(|row| &row.diagnostics).collect::<Vec<_>>());
assert_eq!(result.equity_curve[5].external_cash_flow, 10_000.0);
for ((session, decision), equity) in records.iter().zip(&result.equity_curve) {
assert_eq!(session.date, equity.date);
assert_eq!(session.end_unit_nav.to_bits(), equity.unit_nav.to_bits());
assert!(decision.observed_through.is_none_or(|date| date < session.date));
if decision.observation_count < 10 { assert!(!decision.threshold_breached); }
}
assert!(result.equity_curve[5].unit_nav < result.equity_curve[4].unit_nav);
}
#[test]
fn stock_state_cache_resets_before_reusing_compact_keys_on_another_date() {
let dates = [d(2025, 1, 2), d(2025, 1, 3)];
@@ -3,6 +3,7 @@ use std::collections::{BTreeMap, BTreeSet, HashSet};
use chrono::{NaiveDate, NaiveTime};
use serde::{Deserialize, Serialize};
use serde_json::Value;
use crate::portfolio_loss::PortfolioLossConfig;
use crate::{
DynamicSlippageConfig, MatchingType, PlatformAccountActionKind, PlatformExplicitActionStage,
@@ -915,6 +916,8 @@ pub struct StrategyExpressionRiskConfig {
#[serde(default)]
pub portfolio_drawdown_control: Option<StrategyPortfolioDrawdownControlConfig>,
#[serde(default)]
pub portfolio_loss_control: Option<StrategyPortfolioLossControlConfig>,
#[serde(default)]
pub stop_loss_expr: Option<String>,
#[serde(default)]
pub take_profit_expr: Option<String>,
@@ -963,6 +966,16 @@ pub struct StrategyPortfolioDrawdownControlConfig {
pub cooldown_trading_days: Option<usize>,
}
#[derive(Debug, Clone, Default, Deserialize, Serialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
pub struct StrategyPortfolioLossControlConfig {
pub enabled: Option<bool>,
pub lookback: Option<usize>,
pub loss_trigger: Option<f64>,
pub floor_exposure: Option<f64>,
pub cooldown_trading_days: Option<usize>,
}
#[derive(Debug, Clone, Default, Deserialize, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct StrategyExpressionOrderingConfig {
@@ -2176,6 +2189,18 @@ pub fn platform_expr_config_from_spec(
));
}
}
if let Some(control) = risk.portfolio_loss_control.as_ref()
&& control.enabled.unwrap_or(true)
{
let parsed = PortfolioLossConfig {
lookback: control.lookback.ok_or("portfolioLossControl.lookback is required")?,
loss_trigger: control.loss_trigger.ok_or("portfolioLossControl.lossTrigger is required")?,
floor_exposure: control.floor_exposure.ok_or("portfolioLossControl.floorExposure is required")?,
cooldown_trading_days: control.cooldown_trading_days.ok_or("portfolioLossControl.cooldownTradingDays is required")?,
};
parsed.validate().map_err(|error| error.to_string())?;
cfg.portfolio_loss_control = Some(parsed);
}
if let Some(control) = risk.portfolio_drawdown_control.as_ref()
&& control.enabled.unwrap_or(true)
{
@@ -4564,6 +4589,27 @@ mod tests {
assert_eq!(control.cooldown_trading_days, 30);
}
#[test]
fn portfolio_loss_contract_is_explicit_and_validated() {
let spec = serde_json::json!({"runtimeExpressions":{"risk":{"portfolioLossControl":{
"enabled":true,"lookback":20,"lossTrigger":0.05,"floorExposure":0.2,"cooldownTradingDays":10
}}}});
let cfg = platform_expr_config_from_value("", "", &spec).unwrap();
assert_eq!(cfg.portfolio_loss_control.unwrap(), PortfolioLossConfig {
lookback:20, loss_trigger:0.05, floor_exposure:0.2, cooldown_trading_days:10,
});
for (field, value) in [("lookback", serde_json::json!(0)),
("lossTrigger", serde_json::json!(0.01)), ("floorExposure", serde_json::json!(1.1)),
("cooldownTradingDays", serde_json::json!(0))] {
let mut invalid = spec.clone();
invalid["runtimeExpressions"]["risk"]["portfolioLossControl"][field] = value;
assert!(platform_expr_config_from_value("", "", &invalid).is_err());
}
let mut missing = spec.clone();
missing["runtimeExpressions"]["risk"]["portfolioLossControl"].as_object_mut().unwrap().remove("lossTrigger");
assert!(platform_expr_config_from_value("", "", &missing).is_err());
}
#[test]
fn rejects_invalid_portfolio_drawdown_control() {
let spec = serde_json::json!({
+517
View File
@@ -0,0 +1,517 @@
//! Causal portfolio-loss state, independent of market-data and order adapters.
use std::collections::VecDeque;
use chrono::{DateTime, FixedOffset, NaiveDate, Utc};
use serde::{Deserialize, Serialize};
use thiserror::Error;
const STATE_SCHEMA: &str = "fidc.portfolio-loss-state/v1";
const MAX_OBSERVATIONS: usize = 120;
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
pub struct PortfolioLossConfig {
pub lookback: usize,
pub loss_trigger: f64,
pub floor_exposure: f64,
pub cooldown_trading_days: usize,
}
impl PortfolioLossConfig {
pub fn validate(&self) -> Result<(), PortfolioLossError> {
if !matches!(self.lookback, 10 | 20 | 40 | 60)
|| !self.loss_trigger.is_finite()
|| !(0.02..=0.30).contains(&self.loss_trigger)
|| !self.floor_exposure.is_finite()
|| !(0.0..=1.0).contains(&self.floor_exposure)
|| !(1..=120).contains(&self.cooldown_trading_days)
{
return Err(PortfolioLossError::InvalidConfig);
}
Ok(())
}
}
/// Finalized portfolio accounting, not a market close used as a proxy for NAV.
/// Unit NAV must already exclude external deposits and withdrawals.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
pub struct ClosedPortfolioSession {
pub date: NaiveDate,
pub previous_session_date: Option<NaiveDate>,
pub available_at: DateTime<Utc>,
pub start_unit_nav: f64,
pub end_unit_nav: f64,
pub start_gross_exposure: f64,
pub end_gross_exposure: f64,
pub source_sha256: String,
}
impl ClosedPortfolioSession {
fn validate(&self) -> Result<(), PortfolioLossError> {
let earliest = self.date.and_hms_opt(7, 30, 0).unwrap().and_utc();
if [self.start_unit_nav, self.end_unit_nav]
.iter()
.any(|value| !value.is_finite() || *value <= 0.0)
|| [self.start_gross_exposure, self.end_gross_exposure]
.iter()
.any(|value| !value.is_finite() || *value < 0.0)
|| self
.previous_session_date
.is_some_and(|date| date >= self.date)
|| self.available_at < earliest
|| self.source_sha256.len() != 64
|| !self
.source_sha256
.bytes()
.all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
{
return Err(PortfolioLossError::InvalidObservation);
}
self.unit_return()?;
Ok(())
}
fn unit_return(&self) -> Result<Option<f64>, PortfolioLossError> {
let gross = self.start_gross_exposure.max(self.end_gross_exposure);
if gross <= 1e-12 {
return Ok(None);
}
let value = (self.end_unit_nav / self.start_unit_nav - 1.0) / gross;
if !value.is_finite() {
return Err(PortfolioLossError::InvalidObservation);
}
Ok(Some(value))
}
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
pub struct PortfolioLossDecision {
pub execution_date: NaiveDate,
pub observed_through: Option<NaiveDate>,
pub observation_count: usize,
pub trailing_unit_return: Option<f64>,
pub threshold_breached: bool,
pub newly_triggered: bool,
pub risk_off: bool,
pub cooldown_before: usize,
pub cooldown_after: usize,
pub target_exposure: f64,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
pub struct PortfolioLossState {
schema_version: String,
config: PortfolioLossConfig,
started_on: NaiveDate,
observations: VecDeque<ClosedPortfolioSession>,
last_session: Option<ClosedPortfolioSession>,
cooldown_remaining: usize,
trigger_count: usize,
last_decision: Option<PortfolioLossDecision>,
}
#[derive(Debug, Error, PartialEq, Eq)]
pub enum PortfolioLossError {
#[error("invalid portfolio loss configuration")]
InvalidConfig,
#[error("invalid finalized portfolio session observation")]
InvalidObservation,
#[error("portfolio loss state does not match its frozen configuration")]
StateMismatch,
#[error("portfolio session history is missing, reordered or corrected")]
SessionDiscontinuity,
#[error("portfolio loss observation is not visible at the decision")]
ObservationNotVisible,
#[error("portfolio loss decisions must follow trading-session order")]
DecisionOrder,
}
impl PortfolioLossState {
pub fn new(
config: PortfolioLossConfig,
started_on: NaiveDate,
) -> Result<Self, PortfolioLossError> {
config.validate()?;
Ok(Self {
schema_version: STATE_SCHEMA.to_owned(),
config,
started_on,
observations: VecDeque::new(),
last_session: None,
cooldown_remaining: 0,
trigger_count: 0,
last_decision: None,
})
}
/// Validation is required after deserialization; a JSON hash alone is not
/// account/generation authorization, which belongs to the state owner.
pub fn validate(&self, expected: &PortfolioLossConfig) -> Result<(), PortfolioLossError> {
expected.validate()?;
if self.schema_version != STATE_SCHEMA
|| &self.config != expected
|| self.observations.len() > MAX_OBSERVATIONS
|| self.cooldown_remaining >= expected.cooldown_trading_days
{
return Err(PortfolioLossError::StateMismatch);
}
let mut previous = None;
for item in &self.observations {
item.validate()?;
if item.date < self.started_on
|| previous.is_some_and(|date| item.date <= date)
|| item.unit_return()?.is_none()
{
return Err(PortfolioLossError::StateMismatch);
}
previous = Some(item.date);
}
if let Some(last) = &self.last_session {
last.validate()?;
if last.date < self.started_on
|| previous.is_some_and(|date| date > last.date)
|| (last.unit_return()?.is_some() && self.observations.back() != Some(last))
{
return Err(PortfolioLossError::StateMismatch);
}
} else if !self.observations.is_empty() {
return Err(PortfolioLossError::StateMismatch);
}
if let Some(decision) = &self.last_decision {
let breached = decision
.trailing_unit_return
.is_some_and(|value| value <= -expected.loss_trigger);
let triggered = decision.cooldown_before == 0 && breached;
let after = if decision.cooldown_before > 0 {
decision.cooldown_before - 1
} else if triggered {
expected.cooldown_trading_days - 1
} else {
0
};
if decision.execution_date < self.started_on
|| decision
.observed_through
.is_some_and(|date| date >= decision.execution_date)
|| !decision.target_exposure.is_finite()
|| !(0.0..=1.0).contains(&decision.target_exposure)
|| decision
.trailing_unit_return
.is_some_and(|value| !value.is_finite())
|| decision.cooldown_after != self.cooldown_remaining
|| decision.observation_count > MAX_OBSERVATIONS
|| decision.cooldown_before >= expected.cooldown_trading_days
|| decision.threshold_breached != breached
|| decision.newly_triggered != triggered
|| decision.risk_off != (decision.cooldown_before > 0 || triggered)
|| decision.cooldown_after != after
|| decision.trailing_unit_return.is_some()
!= (decision.observation_count >= expected.lookback)
|| self.trigger_count
> (decision.execution_date - self.started_on).num_days() as usize + 1
{
return Err(PortfolioLossError::StateMismatch);
}
} else if self.cooldown_remaining != 0 || self.trigger_count != 0 {
return Err(PortfolioLossError::StateMismatch);
}
Ok(())
}
/// Exact duplicate delivery is idempotent. Historical corrections require
/// explicit reconciliation instead of changing an already-used window.
pub fn observe(&mut self, session: ClosedPortfolioSession) -> Result<bool, PortfolioLossError> {
self.validate(&self.config)?;
session.validate()?;
if self.last_session.as_ref() == Some(&session) {
return Ok(false);
}
let previous_date = self.last_session.as_ref().map(|value| value.date);
if session.date < self.started_on
|| session.previous_session_date != previous_date
|| previous_date.is_some_and(|date| session.date <= date)
|| (previous_date.is_none() && session.date != self.started_on)
|| self
.last_session
.as_ref()
.is_some_and(|last| session.start_unit_nav != last.end_unit_nav)
{
return Err(PortfolioLossError::SessionDiscontinuity);
}
if session.unit_return()?.is_some() {
self.observations.push_back(session.clone());
if self.observations.len() > MAX_OBSERVATIONS {
self.observations.pop_front();
}
}
self.last_session = Some(session);
Ok(true)
}
pub fn decide(
&mut self,
execution_date: NaiveDate,
previous_completed_session: Option<NaiveDate>,
decision_at: DateTime<Utc>,
risk_on_exposure: f64,
) -> Result<PortfolioLossDecision, PortfolioLossError> {
self.validate(&self.config)?;
if !risk_on_exposure.is_finite() || !(0.0..=1.0).contains(&risk_on_exposure) {
return Err(PortfolioLossError::InvalidConfig);
}
if execution_date < self.started_on
|| previous_completed_session.is_some_and(|date| date >= execution_date)
|| decision_at
.with_timezone(&FixedOffset::east_opt(8 * 3600).unwrap())
.date_naive()
!= execution_date
|| self
.last_decision
.as_ref()
.is_some_and(|last| execution_date < last.execution_date)
{
return Err(PortfolioLossError::DecisionOrder);
}
if let Some(last) = &self.last_session {
if last.date >= execution_date || last.available_at > decision_at {
return Err(PortfolioLossError::ObservationNotVisible);
}
if Some(last.date) != previous_completed_session {
return Err(PortfolioLossError::SessionDiscontinuity);
}
} else if execution_date != self.started_on {
return Err(PortfolioLossError::SessionDiscontinuity);
}
if let Some(cached) = self
.last_decision
.as_mut()
.filter(|last| last.execution_date == execution_date)
{
cached.target_exposure = if cached.risk_off {
self.config.floor_exposure.min(risk_on_exposure)
} else {
risk_on_exposure
};
return Ok(cached.clone());
}
let trailing = if self.observations.len() >= self.config.lookback {
let start = self.observations.len() - self.config.lookback;
let mut growth = 1.0;
for item in self.observations.iter().skip(start) {
growth *=
(1.0 + item.unit_return()?.expect("nonzero exposure observation")).max(0.0);
}
let result = growth - 1.0;
if !result.is_finite() {
return Err(PortfolioLossError::InvalidObservation);
}
Some(result)
} else {
None
};
let breached = trailing.is_some_and(|value| value <= -self.config.loss_trigger);
let before = self.cooldown_remaining;
let triggered = before == 0 && breached;
let risk_off = before > 0 || triggered;
let after = if before > 0 {
before - 1
} else if triggered {
self.config.cooldown_trading_days - 1
} else {
0
};
let decision = PortfolioLossDecision {
execution_date,
observed_through: self.last_session.as_ref().map(|value| value.date),
observation_count: self.observations.len(),
trailing_unit_return: trailing,
threshold_breached: breached,
newly_triggered: triggered,
risk_off,
cooldown_before: before,
cooldown_after: after,
target_exposure: if risk_off {
self.config.floor_exposure.min(risk_on_exposure)
} else {
risk_on_exposure
},
};
self.cooldown_remaining = after;
self.trigger_count += usize::from(triggered);
self.last_decision = Some(decision.clone());
Ok(decision)
}
pub fn last_session(&self) -> Option<&ClosedPortfolioSession> {
self.last_session.as_ref()
}
pub fn last_decision(&self) -> Option<&PortfolioLossDecision> {
self.last_decision.as_ref()
}
}
#[cfg(test)]
mod tests {
use super::*;
use chrono::{Duration, TimeZone};
fn date(day: i64) -> NaiveDate {
NaiveDate::from_ymd_opt(2023, 1, 3).unwrap() + Duration::days(day)
}
fn time(day: i64, hour: u32) -> DateTime<Utc> {
Utc.from_utc_datetime(&date(day).and_hms_opt(hour, 0, 0).unwrap())
}
fn config() -> PortfolioLossConfig {
PortfolioLossConfig {
lookback: 10,
loss_trigger: 0.05,
floor_exposure: 0.2,
cooldown_trading_days: 3,
}
}
fn session(day: i64, start: f64, end: f64, gross: f64) -> ClosedPortfolioSession {
ClosedPortfolioSession {
date: date(day),
previous_session_date: (day > 0).then(|| date(day - 1)),
available_at: time(day, 8),
start_unit_nav: start,
end_unit_nav: end,
start_gross_exposure: gross,
end_gross_exposure: gross,
source_sha256: "a".repeat(64),
}
}
#[test]
fn restart_is_exact_and_duplicate_decisions_do_not_consume_cooldown() {
let mut state = PortfolioLossState::new(config(), date(0)).unwrap();
let mut nav = 1.0;
for day in 0..10 {
let end = nav * 0.994;
state.observe(session(day, nav, end, 1.0)).unwrap();
nav = end;
}
let first = state
.decide(date(10), Some(date(9)), time(10, 1), 0.9)
.unwrap();
assert!(first.newly_triggered);
assert_eq!(first.cooldown_after, 2);
let serialized = serde_json::to_string(&state).unwrap();
let mut restored: PortfolioLossState = serde_json::from_str(&serialized).unwrap();
restored.validate(&config()).unwrap();
assert_eq!(
first,
restored
.decide(date(10), Some(date(9)), time(10, 1), 0.9)
.unwrap()
);
let lowered = restored
.decide(date(10), Some(date(9)), time(10, 2), 0.1)
.unwrap();
assert_eq!(lowered.target_exposure, 0.1);
assert_eq!(lowered.cooldown_after, 2);
for day in 10..15 {
let end = nav * 1.01;
let row = session(day, nav, end, 0.2);
state.observe(row.clone()).unwrap();
restored.observe(row).unwrap();
nav = end;
assert_eq!(
state
.decide(date(day + 1), Some(date(day)), time(day + 1, 1), 0.9)
.unwrap(),
restored
.decide(date(day + 1), Some(date(day)), time(day + 1, 1), 0.9)
.unwrap()
);
}
}
#[test]
fn refuses_future_missing_corrected_and_incomplete_accounting() {
let mut state = PortfolioLossState::new(config(), date(0)).unwrap();
let first = session(0, 1.0, 0.99, 1.0);
assert!(state.observe(first.clone()).unwrap());
assert!(!state.observe(first.clone()).unwrap());
let original = state.clone();
let mut changed = first;
changed.end_unit_nav = 0.98;
assert_eq!(
state.observe(changed),
Err(PortfolioLossError::SessionDiscontinuity)
);
assert_eq!(state, original);
assert_eq!(
state.decide(date(0), None, time(0, 1), 0.9),
Err(PortfolioLossError::ObservationNotVisible)
);
assert_eq!(
state.decide(date(2), Some(date(1)), time(2, 1), 0.9),
Err(PortfolioLossError::SessionDiscontinuity)
);
let mut late = PortfolioLossState::new(config(), date(0)).unwrap();
let mut delayed = session(0, 1.0, 0.99, 1.0);
delayed.available_at = time(2, 1);
late.observe(delayed).unwrap();
assert_eq!(
late.decide(date(1), Some(date(0)), time(1, 1), 0.9),
Err(PortfolioLossError::ObservationNotVisible)
);
let mut invalid = session(1, 0.99, 1.0, 1.0);
invalid.end_unit_nav = f64::NAN;
assert_eq!(
state.observe(invalid),
Err(PortfolioLossError::InvalidObservation)
);
}
#[test]
fn cash_sessions_preserve_continuity_without_inventing_returns() {
let mut state = PortfolioLossState::new(config(), date(0)).unwrap();
for day in 0..20 {
state.observe(session(day, 1.0, 1.0, 0.0)).unwrap();
}
let decision = state
.decide(date(20), Some(date(19)), time(20, 1), 0.9)
.unwrap();
assert_eq!(decision.observation_count, 0);
assert_eq!(decision.trailing_unit_return, None);
assert_eq!(decision.target_exposure, 0.9);
assert_eq!(state.last_session().unwrap().date, date(19));
}
#[test]
fn restored_state_rejects_changed_policy_and_forged_cooldown() {
let state = PortfolioLossState::new(config(), date(0)).unwrap();
let mut changed = config();
changed.floor_exposure = 0.5;
assert_eq!(
state.validate(&changed),
Err(PortfolioLossError::StateMismatch)
);
let mut forged = state.clone();
forged.cooldown_remaining = 1;
assert_eq!(
forged.validate(&config()),
Err(PortfolioLossError::StateMismatch)
);
}
#[test]
fn nav_serialization_preserves_float_bits() {
let mut seed = 0xabcddcba12345678_u64;
for _ in 0..2000 {
seed ^= seed << 13;
seed ^= seed >> 7;
seed ^= seed << 17;
let value = 0.01 + (seed as f64 / u64::MAX as f64) * 9.99;
let serialized = serde_json::to_string(&value).unwrap();
let restored: f64 = serde_json::from_str(&serialized).unwrap();
assert_eq!(value.to_bits(), restored.to_bits());
}
}
}