use std::sync::Arc;
use crate::intervention::TemporalPolicy;
use crate::{Intervention, TargetPopulation, VariableId};
use super::QueryError;
pub const MAX_TEMPORAL_RESPONSE_HORIZONS: usize = 512;
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct TemporalResponseLicense {
pub max_horizons: usize,
pub allowed_policies: &'static [&'static str],
pub default_policy: &'static str,
pub default_treatment_lag: u32,
}
#[derive(Clone, Debug, PartialEq)]
pub struct TemporalResponseSpec {
pub horizons: Arc<[u32]>,
pub policy: TemporalPolicy,
pub max_history_lag: Option<u32>,
}
impl TemporalResponseSpec {
pub const MAX_HORIZONS: usize = MAX_TEMPORAL_RESPONSE_HORIZONS;
pub const POLICY_PULSE: &'static str = "pulse";
pub const POLICY_SUSTAINED: &'static str = "sustained";
pub const ALLOWED_POLICIES: &'static [&'static str] =
&[Self::POLICY_PULSE, Self::POLICY_SUSTAINED];
pub const DEFAULT_POLICY: &'static str = Self::POLICY_PULSE;
pub const DEFAULT_TREATMENT_LAG: u32 = 1;
#[must_use]
pub const fn license() -> TemporalResponseLicense {
TemporalResponseLicense {
max_horizons: Self::MAX_HORIZONS,
allowed_policies: Self::ALLOWED_POLICIES,
default_policy: Self::DEFAULT_POLICY,
default_treatment_lag: Self::DEFAULT_TREATMENT_LAG,
}
}
pub fn parse_policy(tag: &str, at: i32) -> Result<TemporalPolicy, QueryError> {
match tag {
Self::POLICY_PULSE => Ok(TemporalPolicy::pulse(at)),
Self::POLICY_SUSTAINED => Ok(TemporalPolicy::sustained(at, at)),
other => Err(QueryError::InvalidResponse(format!(
"temporal response policy must be {} or {}; got {other}",
Self::POLICY_PULSE,
Self::POLICY_SUSTAINED
))),
}
}
pub fn new(
horizons: impl Into<Arc<[u32]>>,
policy: TemporalPolicy,
max_history_lag: Option<u32>,
) -> Result<Self, QueryError> {
let horizons = horizons.into();
let spec = Self { horizons, policy, max_history_lag };
spec.validate()?;
Ok(spec)
}
pub fn validate(&self) -> Result<(), QueryError> {
if self.horizons.is_empty() {
return Err(QueryError::InvalidResponse(
"temporal response requires at least one horizon".into(),
));
}
if self.horizons.len() > Self::MAX_HORIZONS {
return Err(QueryError::InvalidResponse(
"temporal response horizon count exceeds the materialization cap".into(),
));
}
if self.horizons.iter().any(|h| *h == 0) {
return Err(QueryError::NonPositiveHorizon);
}
if self.horizons.windows(2).any(|w| w[0] >= w[1]) {
return Err(QueryError::InvalidResponse(
"temporal response horizons must be strictly increasing".into(),
));
}
self.policy.validate().map_err(|e| match e {
crate::intervention::InterventionError::InvalidTemporalWindow { from, until } => {
QueryError::InvalidTemporalWindow { from, until }
}
other => QueryError::InvalidIntervention(other.to_string()),
})?;
match &self.policy {
TemporalPolicy::Pulse { .. } | TemporalPolicy::Sustained { .. } => {}
TemporalPolicy::Dynamic { .. } => {
return Err(QueryError::InvalidResponse(
"temporal response policy must be pulse or sustained; \
Dynamic is a TemporalEffect spelling, not a ResponseCurve cell"
.into(),
));
}
}
Ok(())
}
#[must_use]
pub fn max_horizon(&self) -> u32 {
self.horizons.last().copied().unwrap_or(1)
}
pub fn treatment_offset(&self) -> Result<i32, QueryError> {
match &self.policy {
TemporalPolicy::Pulse { at } => Ok(*at),
TemporalPolicy::Sustained { from, .. } => Ok(*from),
TemporalPolicy::Dynamic { active_at, .. } => {
active_at.first().copied().ok_or(QueryError::DynamicPolicyHasNoTreatmentOffset)
}
}
}
}
pub const MAX_NONPARAMETRIC_RESPONSE_DIM: usize = 2;
pub const MAX_MATERIALIZED_GRID_POINTS: usize = 1_000_000;
#[derive(Clone, Debug, PartialEq)]
pub enum GridSpec {
Values(Arc<[f64]>),
Linspace {
start: f64,
end: f64,
points: usize,
},
}
impl GridSpec {
pub fn values(&self) -> Result<Vec<f64>, QueryError> {
self.validate()?;
Ok(match self {
Self::Values(values) => values.to_vec(),
Self::Linspace { start, end, points } => {
let points = u32::try_from(*points).map_err(|_| {
QueryError::InvalidResponse("linspace point count exceeds u32 capacity".into())
})?;
let step = (end - start) / f64::from(points - 1);
(0..points).map(|i| start + f64::from(i) * step).collect()
}
})
}
pub fn validate(&self) -> Result<(), QueryError> {
match self {
Self::Values(values) => {
if values.len() < 2 {
return Err(QueryError::InvalidResponse(
"a response grid requires at least two points".into(),
));
}
if values.iter().any(|v| !v.is_finite()) || values.windows(2).any(|w| w[0] >= w[1])
{
return Err(QueryError::InvalidResponse(
"response-grid values must be finite and strictly increasing".into(),
));
}
}
Self::Linspace { start, end, points } => {
if !start.is_finite() || !end.is_finite() || start >= end || *points < 2 {
return Err(QueryError::InvalidResponse(
"linspace requires finite start < end and at least two points".into(),
));
}
if *points > MAX_MATERIALIZED_GRID_POINTS {
return Err(QueryError::InvalidResponse(
"linspace point count is too large to materialize".into(),
));
}
}
}
Ok(())
}
}
#[derive(Clone, Debug, PartialEq)]
pub struct ContinuousDomain {
pub variable: VariableId,
pub grid: GridSpec,
}
impl ContinuousDomain {
#[must_use]
pub fn new(variable: VariableId, grid: GridSpec) -> Self {
Self { variable, grid }
}
}
#[derive(Clone, Debug, PartialEq)]
pub enum DerivativeWeighting {
Observed,
Uniform {
lower: f64,
upper: f64,
},
Custom(Arc<[f64]>),
}
#[derive(Clone, Copy, Debug, Eq, PartialEq, Hash)]
pub enum DerivativeScale {
Identity,
LogTreatment,
LogOutcome,
LogLog,
}
#[derive(Clone, Debug, Eq, PartialEq, Hash)]
pub enum ObservationSpec {
Complete,
RightCensored {
latent: VariableId,
observed: VariableId,
censoring: VariableId,
event: VariableId,
},
LeftCensored {
latent: VariableId,
observed: VariableId,
censoring: VariableId,
event: VariableId,
},
IntervalCensored {
latent: VariableId,
lower: VariableId,
upper: VariableId,
},
Truncated {
latent: VariableId,
observed: VariableId,
lower: Option<VariableId>,
upper: Option<VariableId>,
},
Selected {
latent: VariableId,
observed: VariableId,
indicator: VariableId,
},
}
#[derive(Clone, Debug, Eq, PartialEq, Hash)]
pub enum ObservationAssumption {
IndependentGiven(Arc<[VariableId]>),
OutcomeIndependentGiven(Arc<[VariableId]>),
Structural(Arc<str>),
}
#[derive(Clone, Debug, PartialEq)]
pub enum ResponseFunctional {
MeanCurve {
outcome: VariableId,
treatment: ContinuousDomain,
},
AverageDerivative {
outcome: VariableId,
treatment: VariableId,
weighting: DerivativeWeighting,
},
PointDerivative {
outcome: VariableId,
treatment: VariableId,
at: f64,
order: u8,
scale: DerivativeScale,
},
DirectionalDerivative {
outcomes: Arc<[VariableId]>,
treatments: Arc<[VariableId]>,
at: Arc<[f64]>,
direction: Arc<[f64]>,
},
Jacobian {
outcomes: Arc<[VariableId]>,
treatments: Arc<[VariableId]>,
at: Arc<[f64]>,
scale: DerivativeScale,
},
InterventionResponse {
outcome: VariableId,
interventions: Arc<[Intervention]>,
},
}
impl ResponseFunctional {
#[must_use]
pub fn treatment_ids(&self) -> Vec<VariableId> {
match self {
Self::MeanCurve { treatment, .. } => vec![treatment.variable],
Self::AverageDerivative { treatment, .. } | Self::PointDerivative { treatment, .. } => {
vec![*treatment]
}
Self::DirectionalDerivative { treatments, .. } | Self::Jacobian { treatments, .. } => {
treatments.to_vec()
}
Self::InterventionResponse { interventions, .. } => {
interventions.iter().filter_map(Intervention::primary_variable).collect()
}
}
}
#[must_use]
pub fn outcome_ids(&self) -> Vec<VariableId> {
match self {
Self::MeanCurve { outcome, .. }
| Self::AverageDerivative { outcome, .. }
| Self::PointDerivative { outcome, .. }
| Self::InterventionResponse { outcome, .. } => vec![*outcome],
Self::DirectionalDerivative { outcomes, .. } | Self::Jacobian { outcomes, .. } => {
outcomes.to_vec()
}
}
}
#[must_use]
pub fn primary_pair(&self) -> Option<(VariableId, VariableId)> {
let treatment = self.treatment_ids().into_iter().next()?;
let outcome = self.outcome_ids().into_iter().next()?;
Some((treatment, outcome))
}
}
#[derive(Clone, Debug, PartialEq)]
pub struct ResponseQuery {
pub functional: ResponseFunctional,
pub target_population: TargetPopulation,
pub observation: ObservationSpec,
pub observation_assumptions: Arc<[ObservationAssumption]>,
pub temporal: Option<TemporalResponseSpec>,
}
impl ResponseQuery {
#[must_use]
pub fn new(functional: ResponseFunctional) -> Self {
Self {
functional,
target_population: TargetPopulation::AllObserved,
observation: ObservationSpec::Complete,
observation_assumptions: Arc::from([]),
temporal: None,
}
}
#[must_use]
pub fn with_observation(
mut self,
observation: ObservationSpec,
assumptions: impl Into<Arc<[ObservationAssumption]>>,
) -> Self {
self.observation = observation;
self.observation_assumptions = assumptions.into();
self
}
#[must_use]
pub fn with_temporal(mut self, temporal: TemporalResponseSpec) -> Self {
self.temporal = Some(temporal);
self
}
#[must_use]
pub const fn is_temporal(&self) -> bool {
self.temporal.is_some()
}
#[must_use]
pub fn with_target_population(mut self, target: TargetPopulation) -> Self {
self.target_population = target;
self
}
pub fn validate(&self) -> Result<(), QueryError> {
self.validate_temporal_attachment()?;
match &self.functional {
ResponseFunctional::MeanCurve { outcome, treatment } => {
if *outcome == treatment.variable {
return Err(QueryError::TreatmentEqualsOutcome { id: *outcome });
}
treatment.grid.validate()?;
}
ResponseFunctional::AverageDerivative { outcome, treatment, weighting } => {
if outcome == treatment {
return Err(QueryError::TreatmentEqualsOutcome { id: *outcome });
}
match weighting {
DerivativeWeighting::Uniform { lower, upper }
if !lower.is_finite() || !upper.is_finite() || lower >= upper =>
{
return Err(QueryError::InvalidResponse(
"uniform derivative weighting requires finite lower < upper".into(),
));
}
DerivativeWeighting::Custom(weights)
if weights.is_empty()
|| weights.iter().any(|w| !w.is_finite() || *w < 0.0)
|| weights.iter().all(|w| *w == 0.0) =>
{
return Err(QueryError::InvalidResponse(
"custom derivative weights must be finite, non-negative, and non-zero"
.into(),
));
}
_ => {}
}
}
ResponseFunctional::PointDerivative { outcome, treatment, at, order, scale } => {
if outcome == treatment {
return Err(QueryError::TreatmentEqualsOutcome { id: *outcome });
}
if !at.is_finite() || !matches!(order, 1 | 2) {
return Err(QueryError::InvalidResponse(
"point derivative requires a finite point and order one or two".into(),
));
}
if matches!(scale, DerivativeScale::LogTreatment | DerivativeScale::LogLog)
&& *at <= 0.0
{
return Err(QueryError::InvalidResponse(
"log-treatment derivative scales require a positive treatment point".into(),
));
}
}
ResponseFunctional::DirectionalDerivative { outcomes, treatments, at, direction } => {
if !response_sets_are_distinct(outcomes, treatments)
|| at.len() != treatments.len()
|| direction.len() != treatments.len()
|| at.iter().chain(direction.iter()).any(|v| !v.is_finite())
|| direction.iter().all(|v| *v == 0.0)
{
return Err(QueryError::InvalidResponse(
"directional derivative dimensions/values are inconsistent".into(),
));
}
}
ResponseFunctional::Jacobian { outcomes, treatments, at, scale } => {
if !response_sets_are_distinct(outcomes, treatments)
|| at.len() != treatments.len()
|| at.iter().any(|v| !v.is_finite())
{
return Err(QueryError::InvalidResponse(
"Jacobian dimensions/values are inconsistent".into(),
));
}
if matches!(scale, DerivativeScale::LogTreatment | DerivativeScale::LogLog)
&& at.iter().any(|v| *v <= 0.0)
{
return Err(QueryError::InvalidResponse(
"log-treatment Jacobians require positive treatment coordinates".into(),
));
}
}
ResponseFunctional::InterventionResponse { outcome, interventions } => {
if interventions.is_empty() {
return Err(QueryError::InvalidResponse(
"intervention response requires at least one intervention".into(),
));
}
for intervention in interventions.iter() {
intervention
.validate()
.map_err(|e| QueryError::InvalidIntervention(e.to_string()))?;
if intervention.primary_variable() == Some(*outcome) {
return Err(QueryError::TreatmentEqualsOutcome { id: *outcome });
}
}
}
}
self.target_population.validate()?;
Ok(())
}
fn validate_temporal_attachment(&self) -> Result<(), QueryError> {
if let Some(temporal) = &self.temporal {
temporal.validate()?;
match &self.functional {
ResponseFunctional::MeanCurve { .. }
| ResponseFunctional::InterventionResponse { .. } => {}
_ => {
return Err(QueryError::InvalidResponse(
"temporal attachment is licensed only for MeanCurve and InterventionResponse"
.into(),
));
}
}
if self.observation != ObservationSpec::Complete {
return Err(QueryError::InvalidResponse(
"temporal response requires complete observation in 0.7".into(),
));
}
}
Ok(())
}
}
fn response_sets_are_distinct(outcomes: &[VariableId], treatments: &[VariableId]) -> bool {
!outcomes.is_empty()
&& !treatments.is_empty()
&& !outcomes.iter().any(|outcome| treatments.contains(outcome))
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn temporal_response_license_is_the_facade_contract() {
let license = TemporalResponseSpec::license();
assert_eq!(license.max_horizons, MAX_TEMPORAL_RESPONSE_HORIZONS);
assert_eq!(license.allowed_policies, TemporalResponseSpec::ALLOWED_POLICIES);
assert_eq!(license.default_policy, TemporalResponseSpec::POLICY_PULSE);
assert!(license.allowed_policies.contains(&license.default_policy));
assert_eq!(license.default_treatment_lag, TemporalResponseSpec::DEFAULT_TREATMENT_LAG);
let at = -i32::try_from(license.default_treatment_lag).unwrap();
assert!(TemporalResponseSpec::parse_policy(license.default_policy, at).is_ok());
assert!(TemporalResponseSpec::parse_policy("dynamic", at).is_err());
let ok: Vec<u32> = (1..=u32::try_from(license.max_horizons).unwrap()).collect();
assert!(TemporalResponseSpec::new(ok, TemporalPolicy::pulse(at), None).is_ok());
let too_many: Vec<u32> = (1..=u32::try_from(license.max_horizons + 1).unwrap()).collect();
assert!(TemporalResponseSpec::new(too_many, TemporalPolicy::pulse(at), None).is_err());
}
#[test]
fn temporal_response_spec_refuses_dynamic_policy() {
let err = TemporalResponseSpec::new(
vec![1u32],
TemporalPolicy::dynamic(crate::DynamicRuleId::from_raw(0), [0]),
None,
)
.unwrap_err();
assert!(matches!(err, QueryError::InvalidResponse(_)));
assert!(err.to_string().contains("pulse or sustained"));
}
#[test]
fn linspace_within_cap_validates_and_materializes() {
let grid = GridSpec::Linspace { start: 0.0, end: 1.0, points: 5 };
assert!(grid.validate().is_ok());
assert_eq!(grid.values().unwrap().len(), 5);
}
#[test]
fn linspace_beyond_materialization_cap_is_rejected() {
let grid =
GridSpec::Linspace { start: 0.0, end: 1.0, points: MAX_MATERIALIZED_GRID_POINTS + 1 };
let err = grid.validate().unwrap_err();
assert!(matches!(err, QueryError::InvalidResponse(_)));
assert!(grid.values().is_err());
}
#[test]
fn response_functional_primary_pair_matches_treatment_and_outcome_ids() {
let treatment = VariableId::from_raw(0);
let outcome = VariableId::from_raw(1);
let functional = ResponseFunctional::AverageDerivative {
outcome,
treatment,
weighting: DerivativeWeighting::Observed,
};
assert_eq!(functional.treatment_ids(), vec![treatment]);
assert_eq!(functional.outcome_ids(), vec![outcome]);
assert_eq!(functional.primary_pair(), Some((treatment, outcome)));
}
#[test]
fn linspace_point_count_far_beyond_u32_capacity_is_still_rejected_by_the_cap() {
let grid = GridSpec::Linspace { start: 0.0, end: 1.0, points: 4_000_000_000 };
let err = grid.validate().unwrap_err();
assert!(matches!(err, QueryError::InvalidResponse(_)));
}
}