1use std::sync::Arc;
6
7use antecedent_core::{
8 AverageEffectQuery, ExecutionContext, KernelPolicy, TemporalEffectQuery, VariableId,
9};
10use antecedent_data::TemporalIndexer;
11use antecedent_data::{
12 DiscoveryEstimationSplit, PanelData, PanelUnit, TableView, TabularData, TimeIndex,
13 TimeSeriesData, ValidityBitmap,
14};
15use antecedent_estimate::{
16 EffectEstimate, EstimationWorkspace, IpwTarget, LinearAdjustmentAte, OverlapPolicy,
17 OverlapReport, TemporalLinearAdjustment,
18};
19use antecedent_identify::IdentifiedEstimand;
20use antecedent_kernels::erfc;
21use antecedent_stats::{
22 FaerBackend, GlmOptions, PropensityFit, PropensityWorkspace, fit_propensity_diagnostic,
23};
24
25use crate::error::ValidationError;
26
27#[derive(Clone, Copy, Debug)]
29pub struct TemporalRefitContext<'a> {
30 pub indexer: &'a TemporalIndexer,
32 pub temporal_query: &'a TemporalEffectQuery,
34 pub split: Option<&'a DiscoveryEstimationSplit>,
36 pub kernel_policy: &'a KernelPolicy,
38 pub time_index: Option<&'a TimeIndex>,
40 pub panel: Option<&'a PanelData>,
42}
43
44impl TemporalRefitContext<'_> {
45 #[must_use]
47 pub fn is_panel(&self) -> bool {
48 self.panel.is_some()
49 }
50}
51
52#[derive(Clone, Debug)]
54#[non_exhaustive]
55pub struct RefutationReport {
56 pub refuter: Arc<str>,
58 pub original_ate: f64,
60 pub refuted_ate: f64,
62 pub comparison: f64,
66 pub informative: bool,
68 pub passed: bool,
70 pub failure_condition: Option<Arc<str>>,
72 pub replicates: u32,
74}
75
76impl RefutationReport {
77 #[must_use]
79 #[allow(clippy::too_many_arguments)]
80 pub fn new(
81 refuter: impl Into<Arc<str>>,
82 original_ate: f64,
83 refuted_ate: f64,
84 comparison: f64,
85 informative: bool,
86 passed: bool,
87 failure_condition: Option<Arc<str>>,
88 replicates: u32,
89 ) -> Self {
90 Self {
91 refuter: refuter.into(),
92 original_ate,
93 refuted_ate,
94 comparison,
95 informative,
96 passed,
97 failure_condition,
98 replicates,
99 }
100 }
101}
102
103#[derive(Clone, Copy, Debug)]
105pub struct RefutationProblem<'a> {
106 pub data: &'a TabularData,
108 pub estimand: &'a IdentifiedEstimand,
110 pub query: &'a AverageEffectQuery,
112 pub original: &'a EffectEstimate,
114 pub estimator: Option<&'a str>,
116 pub temporal: Option<TemporalRefitContext<'a>>,
118}
119
120impl RefutationProblem<'_> {
121 #[must_use]
123 pub fn treatment(&self) -> VariableId {
124 self.query.treatment
125 }
126
127 #[must_use]
129 pub fn outcome(&self) -> VariableId {
130 self.query.outcome
131 }
132}
133
134pub(crate) fn with_replaced_float(
136 data: &TabularData,
137 id: VariableId,
138 values: Arc<[f64]>,
139) -> Result<TabularData, ValidationError> {
140 data.with_replaced_float(id, values).map_err(ValidationError::from)
141}
142
143pub(crate) fn with_extra_float(
145 data: &TabularData,
146 name: &str,
147 values: Arc<[f64]>,
148) -> Result<(TabularData, VariableId), ValidationError> {
149 data.with_appended_float(name, values).map_err(ValidationError::from)
150}
151
152pub(crate) fn fit_once(
154 estimator: &LinearAdjustmentAte,
155 data: &TabularData,
156 estimand: &IdentifiedEstimand,
157 query: &AverageEffectQuery,
158 workspace: &mut EstimationWorkspace,
159 ctx: &ExecutionContext,
160) -> Result<EffectEstimate, ValidationError> {
161 let prep = estimator.prepare(data, estimand, query).map_err(ValidationError::from)?;
162 estimator
163 .fit(&prep, workspace, ctx, antecedent_core::AssumptionSet::new())
164 .map_err(ValidationError::from)
165}
166
167pub(crate) fn refit_effect(
179 problem: &RefutationProblem<'_>,
180 data: &TabularData,
181 estimand: &IdentifiedEstimand,
182 extra_contemporaneous: &[VariableId],
183 caller_estimator: &LinearAdjustmentAte,
184 workspace: &mut EstimationWorkspace,
185 ctx: &ExecutionContext,
186) -> Result<EffectEstimate, ValidationError> {
187 let Some(temporal) = problem.temporal else {
188 let est = forced_refit_estimator(caller_estimator);
189 return fit_once(&est, data, estimand, problem.query, workspace, ctx);
190 };
191 let estimator =
192 TemporalLinearAdjustment::new().with_inner(forced_refit_estimator(caller_estimator));
193 if let Some(panel) = temporal.panel {
194 let rebuilt = panel_from_stacked(panel, data)?;
195 let prep = if extra_contemporaneous.is_empty() {
196 let (prep, _cluster_ids, _panel_times) = estimator
197 .prepare_panel(
198 &rebuilt,
199 estimand,
200 temporal.temporal_query,
201 temporal.indexer,
202 temporal.split,
203 temporal.kernel_policy,
204 )
205 .map_err(ValidationError::from)?;
206 prep
207 } else {
208 panel_prepare_with_extras(
209 &estimator,
210 &rebuilt,
211 estimand,
212 &temporal,
213 extra_contemporaneous,
214 )?
215 };
216 return estimator
217 .fit(&prep, workspace, ctx, antecedent_core::AssumptionSet::new())
218 .map_err(ValidationError::from);
219 }
220 let time_index = temporal.time_index.ok_or(ValidationError::NotApplicable {
221 message: "temporal series refit requires time_index",
222 })?;
223 let series = TimeSeriesData::try_new(data.storage().clone(), time_index.clone())
224 .map_err(ValidationError::from)?;
225 let prep = estimator
226 .prepare_with_extras(
227 &series,
228 estimand,
229 temporal.temporal_query,
230 temporal.indexer,
231 temporal.split,
232 temporal.kernel_policy,
233 extra_contemporaneous,
234 )
235 .map_err(ValidationError::from)?;
236 estimator
237 .fit(&prep, workspace, ctx, antecedent_core::AssumptionSet::new())
238 .map_err(ValidationError::from)
239}
240
241fn panel_prepare_with_extras(
242 estimator: &TemporalLinearAdjustment,
243 panel: &PanelData,
244 estimand: &IdentifiedEstimand,
245 temporal: &TemporalRefitContext<'_>,
246 extra: &[VariableId],
247) -> Result<antecedent_estimate::PreparedEstimationProblem, ValidationError> {
248 let mut all_t = Vec::new();
250 let mut all_y = Vec::new();
251 let mut all_covs: Vec<(VariableId, Vec<f64>)> = Vec::new();
252 let mut adj_keys: Vec<VariableId> = Vec::new();
253 let mut active = 0.0;
254 let mut control = 0.0;
255 let mut treatment_delta = 0.0;
256 let mut first = true;
257 for unit in panel.units() {
258 let prep = estimator
259 .prepare_with_extras(
260 &unit.series,
261 estimand,
262 temporal.temporal_query,
263 temporal.indexer,
264 temporal.split,
265 temporal.kernel_policy,
266 extra,
267 )
268 .map_err(ValidationError::from)?;
269 if first {
270 active = prep.active;
271 control = prep.control;
272 treatment_delta = prep.treatment_delta;
273 adj_keys = prep.adjustment_set.to_vec();
274 all_covs = adj_keys.iter().map(|&id| (id, Vec::new())).collect();
275 first = false;
276 }
277 all_t.extend_from_slice(&prep.treatment);
278 all_y.extend_from_slice(&prep.design.outcome);
279 let nrows = prep.design.nrows;
280 for (i, (_id, dest)) in all_covs.iter_mut().enumerate() {
281 let base = (2 + i) * nrows;
282 dest.extend_from_slice(&prep.design.matrix[base..base + nrows]);
283 }
284 }
285 let cov_refs: Vec<(VariableId, &[f64])> =
286 all_covs.iter().map(|(id, v)| (*id, v.as_slice())).collect();
287 let selected: Vec<usize> = (0..all_t.len()).collect();
288 let design =
289 antecedent_stats::CompiledDesign::linear_adjustment(&all_t, &cov_refs, &all_y, &selected)
290 .map_err(ValidationError::from)?;
291 Ok(antecedent_estimate::PreparedEstimationProblem {
292 design,
293 method: Arc::from("temporal.linear.adjustment.panel"),
294 adjustment_set: Arc::from(adj_keys),
295 overlap: OverlapPolicy::ExplicitOverride,
296 treatment_delta,
297 target_population: antecedent_core::TargetPopulation::AllObserved,
298 treatment: Arc::from(all_t),
299 active,
300 control,
301 })
302}
303
304pub(crate) fn panel_from_stacked(
306 original: &PanelData,
307 stacked: &TabularData,
308) -> Result<PanelData, ValidationError> {
309 let expected = original.total_rows();
310 if stacked.row_count() != expected {
311 return Err(ValidationError::data_msg(format!(
312 "stacked panel refute rows {} != panel total_rows {expected}",
313 stacked.row_count()
314 )));
315 }
316 let mut offset = 0usize;
317 let mut units = Vec::with_capacity(original.unit_count());
318 for u in original.units() {
319 let n = u.series.row_count();
320 let slice = slice_tabular(stacked, offset, n)?;
321 let series =
322 TimeSeriesData::try_new(slice.storage().clone(), u.series.time_index().clone())
323 .map_err(ValidationError::from)?;
324 units.push(PanelUnit { unit_id: u.unit_id, series });
325 offset += n;
326 }
327 PanelData::try_new(Arc::from(units)).map_err(ValidationError::from)
328}
329
330fn slice_tabular(
331 data: &TabularData,
332 start: usize,
333 len: usize,
334) -> Result<TabularData, ValidationError> {
335 use antecedent_data::{Float64Column, OwnedColumn, OwnedColumnarStorage};
336 let storage = data.storage();
337 let end = start + len;
338 if end > data.row_count() {
339 return Err(ValidationError::NotApplicable { message: "panel slice out of range" });
340 }
341 let mut cols = Vec::with_capacity(storage.columns().len());
342 for col in storage.columns() {
343 match col {
344 OwnedColumn::Float64(c) => {
345 let values: Arc<[f64]> = Arc::from(c.values[start..end].to_vec());
346 let validity = ValidityBitmap::all_valid(len);
347 cols.push(OwnedColumn::Float64(
348 Float64Column::new(c.id, values, validity).map_err(ValidationError::from)?,
349 ));
350 }
351 _ => {
352 return Err(ValidationError::NotApplicable {
353 message: "panel refute slice requires float64 columns",
354 });
355 }
356 }
357 }
358 let mask = storage.analysis_mask().map(|m| {
359 let mut bytes = vec![0u8; len.div_ceil(8)];
360 for i in 0..len {
361 if m.is_valid(start + i) {
362 bytes[i / 8] |= 1 << (i % 8);
363 }
364 }
365 ValidityBitmap::from_bytes(bytes, len)
366 });
367 let mask = mask.transpose().map_err(ValidationError::from)?;
368 let weights = storage.weights().map(|w| Arc::<[f64]>::from(w[start..end].to_vec()));
369 let new_storage = OwnedColumnarStorage::try_new(storage.schema().clone(), cols, mask, weights)
370 .map_err(ValidationError::from)?;
371 Ok(TabularData::new(new_storage))
372}
373
374pub fn stack_panel_tabular(panel: &PanelData) -> Result<TabularData, ValidationError> {
376 use antecedent_data::{Float64Column, OwnedColumn, OwnedColumnarStorage};
377 let total = panel.total_rows();
378 let schema = panel.schema().clone();
379 let n_cols = schema.len();
380 let mut col_vals: Vec<Vec<f64>> = (0..n_cols).map(|_| Vec::with_capacity(total)).collect();
381 for u in panel.units() {
382 for (j, dest) in col_vals.iter_mut().enumerate() {
383 let id =
384 VariableId::from_raw(u32::try_from(j).map_err(|_| {
385 ValidationError::data_msg("panel column index exceeds u32::MAX")
386 })?);
387 let vals = u.series.float64_values(id).map_err(ValidationError::from)?;
388 dest.extend_from_slice(&vals);
389 }
390 }
391 let mut cols = Vec::with_capacity(n_cols);
392 for (j, values) in col_vals.into_iter().enumerate() {
393 let id = VariableId::from_raw(
394 u32::try_from(j)
395 .map_err(|_| ValidationError::data_msg("panel column index exceeds u32::MAX"))?,
396 );
397 cols.push(OwnedColumn::Float64(
398 Float64Column::new(id, Arc::from(values), ValidityBitmap::all_valid(total))
399 .map_err(ValidationError::from)?,
400 ));
401 }
402 let storage =
403 OwnedColumnarStorage::try_new(schema, cols, None, None).map_err(ValidationError::from)?;
404 Ok(TabularData::new(storage))
405}
406
407pub(crate) struct DiagnosticPropensityColumns {
409 pub scores: Vec<f64>,
411 pub treatment: Vec<f64>,
413 pub outcome: Option<Vec<f64>>,
415}
416
417pub(crate) fn fit_diagnostic_propensity(
421 problem: &RefutationProblem<'_>,
422 glm_options: &GlmOptions,
423 include_outcome_in_mask: bool,
424 propensity: &mut PropensityWorkspace,
425) -> Result<DiagnosticPropensityColumns, ValidationError> {
426 let mut ids = vec![problem.treatment()];
427 if include_outcome_in_mask {
428 ids.push(problem.outcome());
429 }
430 ids.extend_from_slice(&problem.estimand.adjustment_set);
431 let row_mask = problem.data.complete_case_mask(&ids).map_err(ValidationError::from)?;
432 let treatment = problem
433 .data
434 .float64_masked(problem.treatment(), &row_mask)
435 .map_err(ValidationError::from)?;
436 let outcome = if include_outcome_in_mask {
437 Some(
438 problem
439 .data
440 .float64_masked(problem.outcome(), &row_mask)
441 .map_err(ValidationError::from)?,
442 )
443 } else {
444 None
445 };
446 let nrows = treatment.len();
447 let ncols = 1 + problem.estimand.adjustment_set.len();
448 let mut design = vec![0.0; nrows * ncols];
449 for r in design.iter_mut().take(nrows) {
450 *r = 1.0;
451 }
452 for (i, &z) in problem.estimand.adjustment_set.iter().enumerate() {
453 let col = problem.data.float64_masked(z, &row_mask).map_err(ValidationError::from)?;
454 let base = (1 + i) * nrows;
455 design[base..base + nrows].copy_from_slice(&col);
456 }
457 let backend = FaerBackend;
458 let fit: PropensityFit = fit_propensity_diagnostic(
459 &design,
460 nrows,
461 ncols,
462 &treatment,
463 &backend,
464 propensity,
465 glm_options,
466 )
467 .map_err(ValidationError::from)?;
468 Ok(DiagnosticPropensityColumns { scores: fit.scores, treatment, outcome })
469}
470
471pub(crate) fn diagnostic_overlap_report(
473 problem: &RefutationProblem<'_>,
474 glm_options: &GlmOptions,
475 policy: OverlapPolicy,
476) -> Result<OverlapReport, ValidationError> {
477 let mut ws = PropensityWorkspace::default();
478 diagnostic_overlap_report_with(problem, glm_options, policy, &mut ws)
479}
480
481pub(crate) fn diagnostic_overlap_report_with(
483 problem: &RefutationProblem<'_>,
484 glm_options: &GlmOptions,
485 policy: OverlapPolicy,
486 propensity: &mut PropensityWorkspace,
487) -> Result<OverlapReport, ValidationError> {
488 let cols = fit_diagnostic_propensity(problem, glm_options, false, propensity)?;
489 let weights: Vec<f64> = cols
491 .treatment
492 .iter()
493 .zip(cols.scores.iter())
494 .map(|(&t, &p)| {
495 let p = p.clamp(1e-9, 1.0 - 1e-9);
496 if t > 0.5 { 1.0 / p } else { 1.0 / (1.0 - p) }
497 })
498 .collect();
499 Ok(OverlapReport::from_propensities(
500 &cols.scores,
501 Some(&weights),
502 policy,
503 Some(&cols.treatment),
504 Some(IpwTarget::Ate),
505 None,
506 ))
507}
508
509#[must_use]
511pub(crate) fn linear_estimator_no_bootstrap() -> LinearAdjustmentAte {
512 let mut estimator = LinearAdjustmentAte::new();
513 estimator.bootstrap_replicates = 0;
514 estimator
515}
516
517#[must_use]
521pub(crate) fn forced_refit_estimator(
522 caller_estimator: &LinearAdjustmentAte,
523) -> LinearAdjustmentAte {
524 let mut estimator = caller_estimator.clone();
525 estimator.bootstrap_replicates = 0;
526 estimator.overlap = OverlapPolicy::ExplicitOverride;
527 estimator
528}
529
530#[derive(Clone, Copy, Debug, Eq, PartialEq)]
532pub(crate) enum NoiseReplaceTarget {
533 Treatment,
535 Outcome,
537}
538
539#[allow(clippy::too_many_arguments)]
544pub(crate) fn noise_replace_refute(
545 problem: &RefutationProblem<'_>,
546 workspace: &mut EstimationWorkspace,
547 ctx: &ExecutionContext,
548 estimator: &LinearAdjustmentAte,
549 replicates: u32,
550 alpha: f64,
551 target: NoiseReplaceTarget,
552 stream_base: u64,
553 refuter_id: &'static str,
554 failure_label: &'static str,
555) -> Result<RefutationReport, ValidationError> {
556 if replicates < 2 {
557 return Err(ValidationError::NotApplicable {
558 message: "noise-replace refuter requires replicates >= 2",
559 });
560 }
561 let replace_id = match target {
562 NoiseReplaceTarget::Treatment => problem.treatment(),
563 NoiseReplaceTarget::Outcome => problem.outcome(),
564 };
565 let n = problem.data.row_count();
566 let mut ates = Vec::with_capacity(replicates as usize);
567 for r in 0..replicates {
568 let mut noise = vec![0.0; n];
569 fill_gaussian(&mut noise, ctx, stream_base.wrapping_add(u64::from(r)));
570 let data = with_replaced_float(problem.data, replace_id, Arc::from(noise))?;
571 let est = refit_effect(problem, &data, problem.estimand, &[], estimator, workspace, ctx)?;
572 ates.push(est.ate);
573 }
574 let mean_ate = ates.iter().sum::<f64>() / f64::from(replicates);
575 let p_value = replicate_p_value(&ates, 0.0);
576 let passed = p_value >= alpha;
577 Ok(RefutationReport {
578 refuter: Arc::from(refuter_id),
579 original_ate: problem.original.ate,
580 refuted_ate: mean_ate,
581 comparison: p_value,
582 informative: true,
583 passed,
584 failure_condition: (!passed).then(|| {
585 Arc::from(format!(
586 "{failure_label} ATE distribution (mean {mean_ate}) is inconsistent with zero \
587 (p={p_value} < alpha={alpha})"
588 ))
589 }),
590 replicates,
591 })
592}
593
594pub(crate) fn replicate_p_value(samples: &[f64], hypothesized: f64) -> f64 {
598 if samples.len() < 2 {
599 return 1.0;
600 }
601 #[allow(clippy::cast_precision_loss)]
602 let mean = samples.iter().sum::<f64>() / samples.len() as f64;
603 let sd = sample_sd(samples);
604 let scale = mean.abs().max(hypothesized.abs()).max(1.0);
605 if !sd.is_finite() {
606 return 1.0;
607 }
608 if sd <= 1e-12 * scale {
609 return if (hypothesized - mean).abs() <= 1e-9 * scale { 1.0 } else { 0.0 };
610 }
611 let z = (hypothesized - mean) / sd;
612 if !z.is_finite() {
613 return 1.0;
619 }
620 erfc(z.abs() / std::f64::consts::SQRT_2).clamp(0.0, 1.0)
624}
625
626pub(crate) fn float64_full(
628 data: &TabularData,
629 id: VariableId,
630) -> Result<Vec<f64>, ValidationError> {
631 data.float64_values(id).map_err(ValidationError::from)
632}
633
634pub(crate) fn with_row_subset(
637 data: &TabularData,
638 keep_fraction: f64,
639 ctx: &ExecutionContext,
640 stream_id: u64,
641) -> Result<TabularData, ValidationError> {
642 let n = data.row_count();
643 let mut rng = ctx.rng.stream(stream_id);
644 let mut bytes = vec![0u8; n.div_ceil(8)];
645 for i in 0..n {
646 if rng.next_f64() < keep_fraction {
647 bytes[i / 8] |= 1 << (i % 8);
648 }
649 }
650 let mask = ValidityBitmap::from_bytes(bytes, n).map_err(ValidationError::from)?;
651 data.with_analysis_mask(mask).map_err(ValidationError::from)
652}
653
654pub(crate) fn with_resampled_rows(
660 data: &TabularData,
661 resample_ids: &[VariableId],
662 row_idx: &[usize],
663 keep: &[bool],
664) -> Result<TabularData, ValidationError> {
665 let mut out = data.clone();
666 for &id in resample_ids {
667 let full = float64_full(&out, id)?;
668 let resampled: Vec<f64> = row_idx.iter().map(|&i| full[i]).collect();
669 out = with_replaced_float(&out, id, Arc::from(resampled))?;
670 }
671 if keep.iter().all(|&k| k) {
672 return Ok(out);
673 }
674 let n = keep.len();
675 let mut bytes = vec![0u8; n.div_ceil(8)];
676 for (i, &k) in keep.iter().enumerate() {
677 if k {
678 bytes[i / 8] |= 1 << (i % 8);
679 }
680 }
681 let mask = ValidityBitmap::from_bytes(bytes, n).map_err(ValidationError::from)?;
682 out.with_analysis_mask(mask).map_err(ValidationError::from)
683}
684
685pub(crate) fn complete_case_rows(
687 data: &TabularData,
688 ids: &[VariableId],
689) -> Result<(Vec<bool>, Vec<usize>), ValidationError> {
690 let mask = data.complete_case_mask(ids).map_err(ValidationError::from)?;
691 let valid: Vec<usize> = mask.iter().enumerate().filter_map(|(i, &k)| k.then_some(i)).collect();
692 Ok((mask, valid))
693}
694
695pub(crate) fn masked_sample_sd(
697 data: &TabularData,
698 id: VariableId,
699 mask: &[bool],
700) -> Result<f64, ValidationError> {
701 let vals = data.float64_masked(id, mask).map_err(ValidationError::from)?;
702 Ok(sample_sd(&vals))
703}
704
705pub(crate) fn sample_sd(values: &[f64]) -> f64 {
707 antecedent_stats::sample_std(values)
708}
709
710pub(crate) fn fill_gaussian(out: &mut [f64], ctx: &ExecutionContext, stream_id: u64) {
715 let mut rng = ctx.rng.stream(stream_id);
716 antecedent_kernels::fill_standard_normal(&mut rng, out);
717}
718
719#[cfg(test)]
720mod tests {
721 use super::*;
722
723 #[test]
724 fn replicate_p_value_zero_sd_matches_hypothesis_is_valid_probability() {
725 let samples = [1.5, 1.5, 1.5, 1.5];
728 let p = replicate_p_value(&samples, 1.5);
729 assert!((0.0..=1.0).contains(&p), "p_value {p} outside [0, 1]");
730 assert!(!p.is_nan(), "p_value must not be NaN");
731 }
732
733 #[test]
734 fn replicate_p_value_zero_sd_away_from_hypothesis_is_valid_probability() {
735 let samples = [1.5, 1.5, 1.5, 1.5];
737 let p = replicate_p_value(&samples, 0.0);
738 assert!((0.0..=1.0).contains(&p), "p_value {p} outside [0, 1]");
739 assert!(!p.is_nan(), "p_value must not be NaN");
740 }
741
742 #[test]
743 fn replicate_p_value_nan_mean_does_not_propagate_nan() {
744 let samples = [1.0, 2.0, f64::NAN, 3.0];
749 let p = replicate_p_value(&samples, 0.0);
750 assert!((0.0..=1.0).contains(&p), "p_value {p} outside [0, 1]");
751 assert!(!p.is_nan(), "p_value must not be NaN");
752 }
753}