1#![allow(
6 clippy::cast_possible_truncation,
7 clippy::cast_precision_loss,
8 clippy::cast_sign_loss,
9 clippy::many_single_char_names,
10 clippy::needless_range_loop,
11 clippy::too_many_arguments
12)]
13
14use std::sync::Arc;
15
16use antecedent_core::{CausalRng, ExecutionContext, Intervention, VariableId};
17use antecedent_data::{TableView, TabularData};
18use antecedent_kernels::standard_normal;
19
20use crate::batch::{MechanismWorkspace, ParentBatch};
21use crate::compile::CompiledCausalModel;
22use crate::error::ModelError;
23use crate::mechanism::log_prob_column;
24use crate::sample::sample_interventional;
25
26#[derive(Clone, Debug)]
28pub struct DoSampleResult {
29 pub values: Arc<[f64]>,
31 pub weights: Arc<[f64]>,
33 pub method: Arc<str>,
35 pub notes: Vec<Arc<str>>,
37 pub accept_rate: Option<f64>,
39 pub bandwidth: Option<f64>,
41 pub ess: Option<f64>,
50}
51
52#[derive(Clone, Debug)]
63pub struct WeightingDoSampler {
64 pub treatment: VariableId,
66 pub outcome: VariableId,
68}
69
70impl WeightingDoSampler {
71 #[must_use]
73 pub fn new(treatment: VariableId, outcome: VariableId) -> Self {
74 Self { treatment, outcome }
75 }
76
77 pub fn estimate(
83 &self,
84 model: &CompiledCausalModel,
85 data: &TabularData,
86 treatment_value: f64,
87 _ctx: &ExecutionContext,
88 ) -> Result<DoSampleResult, ModelError> {
89 let t_dense = model
90 .dense_of(self.treatment)
91 .ok_or_else(|| ModelError::Shape { message: "treatment not in model".into() })?;
92 let y = data.float64_cow(self.outcome).map_err(ModelError::from)?;
93 let t = data.float64_cow(self.treatment).map_err(ModelError::from)?;
94 let n = y.len();
95 let gather = model
96 .gather_for(t_dense)
97 .ok_or_else(|| ModelError::Shape { message: "missing gather for treatment".into() })?;
98
99 let mut weights = vec![0.0; n];
100 let mut values = Vec::with_capacity(n);
101 let mut notes = Vec::new();
102
103 if gather.n_parents() == 0 {
104 let mut selected = 0usize;
106 for i in 0..n {
107 if (t[i] - treatment_value).abs() < 1e-9 {
108 values.push(y[i]);
109 weights[selected] = 1.0;
110 selected += 1;
111 }
112 }
113 weights.truncate(selected);
114 if selected == 0 {
115 return Err(ModelError::Numerical {
116 message: "weighting sampler: no observational units match do-value".into(),
117 });
118 }
119 notes.push(Arc::from("root treatment: exact match reweighting"));
120 return Ok(DoSampleResult {
121 values: Arc::from(values),
122 weights: Arc::from(weights),
123 method: Arc::from("do_weighting"),
124 notes,
125 accept_rate: None,
126 bandwidth: None,
127 ess: None,
128 });
129 }
130
131 let slot = model.mechanisms.get(t_dense);
133 let mut parent_cols = Vec::new();
134 for &p in gather.parents.iter() {
135 let var = model.output_layout.variables[p.as_usize()];
136 parent_cols.push(data.float64_cow(var).map_err(ModelError::from)?);
137 }
138 let n_par = gather.n_parents();
139 let mut parent_mat = vec![0.0; n * n_par];
140 for (pi, col) in parent_cols.iter().enumerate() {
141 for r in 0..n {
142 parent_mat[pi * n + r] = col[r];
143 }
144 }
145 let parents = ParentBatch { n_rows: n, n_parents: n_par, values: &parent_mat };
146 let mut lp_obs = vec![0.0; n];
147 let has_density = log_prob_column(slot, &t, parents, &mut lp_obs).is_ok();
148 let bw = silverman_bandwidth(&t).max(1e-8);
149 let inv_norm = 1.0 / (bw * (2.0 * std::f64::consts::PI).sqrt());
150 for i in 0..n {
151 let z = (t[i] - treatment_value) / bw;
152 let kernel = inv_norm * (-0.5 * z * z).exp();
153 let w = if has_density && lp_obs[i].is_finite() {
154 let dens = lp_obs[i].exp().max(1e-300);
155 (kernel / dens).min(1e6)
156 } else {
157 kernel
158 };
159 weights[i] = w;
160 values.push(y[i]);
161 }
162 let wsum: f64 = weights.iter().sum();
163 if wsum.partial_cmp(&0.0) != Some(std::cmp::Ordering::Greater) {
164 return Err(ModelError::Numerical {
165 message: "weighting sampler: zero weight mass".into(),
166 });
167 }
168 for w in &mut weights {
169 *w /= wsum;
170 }
171 let sum_sq: f64 = weights.iter().map(|w| w * w).sum();
180 let ess = if sum_sq > 0.0 { 1.0 / sum_sq } else { 0.0 };
181 let min_ess_floor = (n as f64 * 0.05).max(2.0);
187 if ess < min_ess_floor {
188 notes.push(Arc::from(format!(
189 "degenerate weighting: ess={ess:.3} of n={n} (< floor {min_ess_floor:.3}); \
190 estimate is dominated by a handful of units"
191 )));
192 }
193 notes.push(Arc::from(format!("IPW / Silverman-kernel weighting (bandwidth={bw:.6})")));
194 Ok(DoSampleResult {
195 values: Arc::from(values),
196 weights: Arc::from(weights),
197 method: Arc::from("do_weighting"),
198 notes,
199 accept_rate: None,
200 bandwidth: Some(bw),
201 ess: Some(ess),
202 })
203 }
204
205 #[must_use]
207 pub fn weighted_mean(result: &DoSampleResult) -> f64 {
208 if result.values.is_empty() {
209 return f64::NAN;
210 }
211 if result.weights.is_empty() {
212 return result.values.iter().sum::<f64>() / result.values.len() as f64;
213 }
214 let wsum: f64 = result.weights.iter().sum();
215 if wsum.partial_cmp(&0.0) != Some(std::cmp::Ordering::Greater) {
216 return f64::NAN;
217 }
218 result.values.iter().zip(result.weights.iter()).map(|(v, w)| v * w).sum::<f64>() / wsum
219 }
220}
221
222#[derive(Clone, Debug)]
224pub struct KdeDoSampler {
225 pub outcome: VariableId,
227 pub bandwidth: Option<f64>,
229}
230
231impl KdeDoSampler {
232 #[must_use]
234 pub fn new(outcome: VariableId) -> Self {
235 Self { outcome, bandwidth: None }
236 }
237
238 pub fn sample(
244 &self,
245 model: &CompiledCausalModel,
246 interventions: &[Intervention],
247 n_draws: usize,
248 rng: &mut CausalRng,
249 ws: &mut MechanismWorkspace,
250 ctx: &ExecutionContext,
251 ) -> Result<DoSampleResult, ModelError> {
252 let batch = sample_interventional(model, interventions, n_draws, rng, ws, ctx)?;
253 let dense = model
254 .dense_of(self.outcome)
255 .ok_or_else(|| ModelError::Shape { message: "outcome not in model".into() })?;
256 let col = batch.column(dense.as_usize())?;
257 let bw = self.bandwidth.unwrap_or_else(|| silverman_bandwidth(col));
258 Ok(DoSampleResult {
259 values: Arc::from(col.to_vec()),
260 weights: Arc::from(vec![1.0 / n_draws as f64; n_draws]),
261 method: Arc::from("do_kde"),
262 notes: Vec::new(),
263 accept_rate: None,
264 bandwidth: Some(bw),
265 ess: None,
266 })
267 }
268
269 #[must_use]
271 pub fn density(result: &DoSampleResult, x: f64) -> f64 {
272 let bw = result.bandwidth.unwrap_or(1.0).max(1e-8);
273 let n = result.values.len() as f64;
274 let inv = 1.0 / (bw * (2.0 * std::f64::consts::PI).sqrt());
275 let mut dens = 0.0;
276 for &v in result.values.iter() {
277 let z = (x - v) / bw;
278 dens += inv * (-0.5 * z * z).exp();
279 }
280 dens / n.max(1.0)
281 }
282}
283
284fn silverman_bandwidth(x: &[f64]) -> f64 {
285 let n = x.len() as f64;
286 if n < 2.0 {
287 return 1.0;
288 }
289 let mean = x.iter().sum::<f64>() / n;
290 let var = x.iter().map(|v| (v - mean).powi(2)).sum::<f64>() / (n - 1.0);
291 let sd = var.sqrt().max(1e-8);
292 1.06 * sd * n.powf(-0.2)
293}
294
295#[derive(Clone, Debug)]
303pub struct McmcDoSampler {
304 pub outcome: VariableId,
306 pub proposal_sd: f64,
308 pub burn_in: usize,
310 pub thin: usize,
312}
313
314impl Default for McmcDoSampler {
315 fn default() -> Self {
316 Self { outcome: VariableId::from_raw(0), proposal_sd: 0.5, burn_in: 100, thin: 2 }
317 }
318}
319
320impl McmcDoSampler {
321 #[must_use]
323 pub fn new(outcome: VariableId) -> Self {
324 Self { outcome, ..Self::default() }
325 }
326
327 pub fn sample(
333 &self,
334 model: &CompiledCausalModel,
335 interventions: &[Intervention],
336 n_samples: usize,
337 rng: &mut CausalRng,
338 ws: &mut MechanismWorkspace,
339 ctx: &ExecutionContext,
340 ) -> Result<DoSampleResult, ModelError> {
341 let pilot = sample_interventional(model, interventions, n_samples.max(64), rng, ws, ctx)?;
342 let dense = model
343 .dense_of(self.outcome)
344 .ok_or_else(|| ModelError::Shape { message: "outcome not in model".into() })?;
345 let pilot_col = pilot.column(dense.as_usize())?;
346 let pilot_bw = silverman_bandwidth(pilot_col);
347 let kde = DoSampleResult {
348 values: Arc::from(pilot_col.to_vec()),
349 weights: Arc::from([]),
350 method: Arc::from("pilot"),
351 notes: Vec::new(),
352 accept_rate: None,
353 bandwidth: Some(pilot_bw),
354 ess: None,
355 };
356
357 let mut current = pilot_col[0];
358 let mut accepted = 0usize;
359 let mut total = 0usize;
360 let mut out = Vec::with_capacity(n_samples);
361 let iters = self.burn_in + n_samples * self.thin.max(1);
362 let degenerate =
365 pilot_bw < 1e-6 || pilot_col.iter().all(|&v| (v - pilot_col[0]).abs() < 1e-12);
366
367 let mut p_cur =
371 if degenerate { f64::NAN } else { KdeDoSampler::density(&kde, current).max(1e-300) };
372 for i in 0..iters {
373 if degenerate {
374 let idx = (rng.next_f64() * pilot_col.len() as f64).floor() as usize
375 % pilot_col.len().max(1);
376 current = pilot_col[idx];
377 accepted += 1;
378 total += 1;
379 } else {
380 let z = standard_normal(rng);
381 let prop = current + self.proposal_sd * z;
382 let p_prop = KdeDoSampler::density(&kde, prop).max(1e-300);
383 let accept = (p_prop / p_cur).min(1.0);
384 total += 1;
385 if rng.next_f64() < accept {
386 current = prop;
387 p_cur = p_prop;
388 accepted += 1;
389 }
390 }
391 if i >= self.burn_in && (i - self.burn_in) % self.thin.max(1) == 0 {
392 out.push(current);
393 if out.len() == n_samples {
394 break;
395 }
396 }
397 }
398 let rate = accepted as f64 / total.max(1) as f64;
399 Ok(DoSampleResult {
400 values: Arc::from(out),
401 weights: Arc::from([]),
402 method: Arc::from("do_mcmc"),
403 notes: vec![Arc::from(format!("mh_accept_rate={rate}"))],
404 accept_rate: Some(rate),
405 bandwidth: kde.bandwidth,
406 ess: None,
407 })
408 }
409}
410
411pub fn interventional_mean(
417 model: &CompiledCausalModel,
418 interventions: &[Intervention],
419 outcome: VariableId,
420 n_draws: usize,
421 rng: &mut CausalRng,
422 ws: &mut MechanismWorkspace,
423 ctx: &ExecutionContext,
424) -> Result<f64, ModelError> {
425 let batch = sample_interventional(model, interventions, n_draws, rng, ws, ctx)?;
426 let dense = model
427 .dense_of(outcome)
428 .ok_or_else(|| ModelError::Shape { message: "outcome not in model".into() })?;
429 let col = batch.column(dense.as_usize())?;
430 Ok(col.iter().sum::<f64>() / col.len().max(1) as f64)
431}
432
433#[cfg(test)]
434mod tests {
435 use super::*;
436 use crate::registry::{MechanismRegistry, SelectionPolicy};
437 use antecedent_core::{
438 CausalSchemaBuilder, MeasurementSpec, RoleHint, SmallRoleSet, Value, ValueType,
439 };
440 use antecedent_data::column::{Float64Column, ValidityBitmap};
441 use antecedent_data::{OwnedColumn, OwnedColumnarStorage};
442 use antecedent_graph::{Dag, DenseNodeId};
443
444 fn binary_treatment_scm() -> (CompiledCausalModel, TabularData) {
445 let n = 80usize;
446 let mut b = CausalSchemaBuilder::new();
447 b.add_variable(
448 "t",
449 ValueType::Continuous,
450 SmallRoleSet::from_hint(RoleHint::TreatmentCandidate),
451 None,
452 None,
453 MeasurementSpec::default(),
454 )
455 .unwrap();
456 b.add_variable(
457 "y",
458 ValueType::Continuous,
459 SmallRoleSet::from_hint(RoleHint::OutcomeCandidate),
460 None,
461 None,
462 MeasurementSpec::default(),
463 )
464 .unwrap();
465 let schema = b.build().unwrap();
466 let mut t = vec![0.0; n];
467 let mut y = vec![0.0; n];
468 for i in 0..n {
469 t[i] = if i % 2 == 0 { 1.0 } else { 0.0 };
470 y[i] = 2.0 * t[i];
471 }
472 let validity = ValidityBitmap::all_valid(n);
473 let cols = vec![
474 OwnedColumn::Float64(
475 Float64Column::new(VariableId::from_raw(0), Arc::from(t), validity.clone())
476 .unwrap(),
477 ),
478 OwnedColumn::Float64(
479 Float64Column::new(VariableId::from_raw(1), Arc::from(y), validity).unwrap(),
480 ),
481 ];
482 let data =
483 TabularData::new(OwnedColumnarStorage::try_new(schema, cols, None, None).unwrap());
484 let mut g = Dag::with_variables(2);
485 g.insert_directed(DenseNodeId::from_raw(0), DenseNodeId::from_raw(1)).unwrap();
486 let compiled = CompiledCausalModel::compile(g).unwrap();
487 let (store, _) = MechanismRegistry::standard()
488 .assign_and_fit(&compiled, &data, SelectionPolicy::BestScore)
489 .unwrap();
490 (compiled.with_mechanisms(store), data)
491 }
492
493 #[test]
494 fn weighting_recovers_treated_mean() {
495 let (model, data) = binary_treatment_scm();
496 let ctx = ExecutionContext::for_tests(1);
497 let sampler = WeightingDoSampler::new(VariableId::from_raw(0), VariableId::from_raw(1));
498 let res = sampler.estimate(&model, &data, 1.0, &ctx).unwrap();
499 let mean = WeightingDoSampler::weighted_mean(&res);
500 assert!((mean - 2.0).abs() < 1e-9, "mean={mean}");
501 }
502
503 fn confounded_continuous_scm(n: usize) -> (CompiledCausalModel, TabularData) {
507 let mut b = CausalSchemaBuilder::new();
508 for name in ["z", "t", "y"] {
509 b.add_variable(
510 name,
511 ValueType::Continuous,
512 SmallRoleSet::from_hint(RoleHint::OutcomeCandidate),
513 None,
514 None,
515 MeasurementSpec::default(),
516 )
517 .unwrap();
518 }
519 let schema = b.build().unwrap();
520 let mut z = vec![0.0; n];
521 let mut t = vec![0.0; n];
522 let mut y = vec![0.0; n];
523 for i in 0..n {
524 z[i] = (i as f64 - (n as f64) / 2.0) * (10.0 / n as f64);
527 t[i] = z[i] + 0.1 * (i as f64 * 0.7).sin();
528 y[i] = t[i] + 0.5 * z[i] + 0.05 * (i as f64 * 0.3).cos();
529 }
530 let validity = ValidityBitmap::all_valid(n);
531 let cols = vec![
532 OwnedColumn::Float64(
533 Float64Column::new(VariableId::from_raw(0), Arc::from(z), validity.clone())
534 .unwrap(),
535 ),
536 OwnedColumn::Float64(
537 Float64Column::new(VariableId::from_raw(1), Arc::from(t), validity.clone())
538 .unwrap(),
539 ),
540 OwnedColumn::Float64(
541 Float64Column::new(VariableId::from_raw(2), Arc::from(y), validity).unwrap(),
542 ),
543 ];
544 let data =
545 TabularData::new(OwnedColumnarStorage::try_new(schema, cols, None, None).unwrap());
546 let mut g = Dag::with_variables(3);
547 g.insert_directed(DenseNodeId::from_raw(0), DenseNodeId::from_raw(1)).unwrap();
548 g.insert_directed(DenseNodeId::from_raw(0), DenseNodeId::from_raw(2)).unwrap();
549 g.insert_directed(DenseNodeId::from_raw(1), DenseNodeId::from_raw(2)).unwrap();
550 let compiled = CompiledCausalModel::compile(g).unwrap();
551 let (store, _) = MechanismRegistry::standard()
552 .assign_and_fit(&compiled, &data, SelectionPolicy::BestScore)
553 .unwrap();
554 (compiled.with_mechanisms(store), data)
555 }
556
557 #[test]
564 fn weighting_ess_flags_tail_degeneracy() {
565 let n = 200usize;
566 let (model, data) = confounded_continuous_scm(n);
567 let ctx = ExecutionContext::for_tests(1);
568 let sampler = WeightingDoSampler::new(VariableId::from_raw(1), VariableId::from_raw(2));
569
570 let healthy = sampler.estimate(&model, &data, 0.0, &ctx).unwrap();
572 let healthy_ess = healthy.ess.expect("weighted branch must report ess");
573
574 let degenerate = sampler.estimate(&model, &data, 20.0, &ctx).unwrap();
578 let degenerate_ess = degenerate.ess.expect("weighted branch must report ess");
579
580 assert!(
581 healthy_ess > 0.2 * n as f64,
582 "expected healthy ess to be a healthy fraction of n={n}, got {healthy_ess}"
583 );
584 assert!(
585 degenerate_ess < 0.05 * n as f64,
586 "expected degenerate ess << n={n}, got {degenerate_ess}"
587 );
588 assert!(
589 degenerate_ess < healthy_ess,
590 "degenerate ess ({degenerate_ess}) should be far below healthy ess ({healthy_ess})"
591 );
592 assert!(
593 degenerate.notes.iter().any(|n| n.contains("degenerate weighting")),
594 "expected a degeneracy note, got {:?}",
595 degenerate.notes
596 );
597 assert!(
598 !healthy.notes.iter().any(|n| n.contains("degenerate weighting")),
599 "healthy case should not be flagged as degenerate, got {:?}",
600 healthy.notes
601 );
602 }
603
604 #[test]
605 fn kde_and_mcmc_run() {
606 let n = 40;
607 let mut b = CausalSchemaBuilder::new();
608 b.add_variable(
609 "t",
610 ValueType::Continuous,
611 SmallRoleSet::from_hint(RoleHint::TreatmentCandidate),
612 None,
613 None,
614 MeasurementSpec::default(),
615 )
616 .unwrap();
617 b.add_variable(
618 "y",
619 ValueType::Continuous,
620 SmallRoleSet::from_hint(RoleHint::OutcomeCandidate),
621 None,
622 None,
623 MeasurementSpec::default(),
624 )
625 .unwrap();
626 let schema = b.build().unwrap();
627 let mut t = vec![0.0; n];
628 let mut y = vec![0.0; n];
629 for i in 0..n {
630 t[i] = if i % 2 == 0 { 1.0 } else { 0.0 };
631 y[i] = 2.0 * t[i] + 0.05 * ((i as f64) - 20.0);
633 }
634 let validity = ValidityBitmap::all_valid(n);
635 let cols = vec![
636 OwnedColumn::Float64(
637 Float64Column::new(VariableId::from_raw(0), Arc::from(t), validity.clone())
638 .unwrap(),
639 ),
640 OwnedColumn::Float64(
641 Float64Column::new(VariableId::from_raw(1), Arc::from(y), validity).unwrap(),
642 ),
643 ];
644 let data =
645 TabularData::new(OwnedColumnarStorage::try_new(schema, cols, None, None).unwrap());
646 let mut g = Dag::with_variables(2);
647 g.insert_directed(DenseNodeId::from_raw(0), DenseNodeId::from_raw(1)).unwrap();
648 let compiled = CompiledCausalModel::compile(g).unwrap();
649 let (store, _) = MechanismRegistry::standard()
650 .assign_and_fit(&compiled, &data, SelectionPolicy::BestScore)
651 .unwrap();
652 let model = compiled.with_mechanisms(store);
653
654 let ctx = ExecutionContext::for_tests(1);
655 let mut rng = CausalRng::from_seed(3);
656 let mut ws = MechanismWorkspace::default();
657 let iv = [Intervention::set(VariableId::from_raw(0), Value::f64(1.0))];
658 let kde = KdeDoSampler::new(VariableId::from_raw(1))
659 .sample(&model, &iv, 40, &mut rng, &mut ws, &ctx)
660 .unwrap();
661 assert_eq!(kde.values.len(), 40);
662 let mcmc = McmcDoSampler::new(VariableId::from_raw(1))
663 .sample(&model, &iv, 30, &mut rng, &mut ws, &ctx)
664 .unwrap();
665 assert_eq!(mcmc.values.len(), 30);
666 assert!(mcmc.accept_rate.unwrap() > 0.0);
667 }
668}