1use ndarray::{concatenate, Array1, ArrayView1, ArrayView2, Axis};
21
22use super::quantile::half_split;
23use super::{dml_model_api, LinearScoreModel, NuisanceTarget, ScoreEstimate, ScoreInput};
24use crate::crossfit::{cross_fit_predict_proba, cross_fit_predict_subset, Fold, SampleSplitting};
25use crate::data::{gather, is_binary, DmlData};
26use crate::error::{DmlError, Result};
27use crate::learner::{Classifier, Regressor};
28use crate::score::LinearScore;
29
30#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
32pub enum SsmScore {
33 #[default]
35 MissingAtRandom,
36 Nonignorable,
38}
39
40pub struct Ssm {
54 ml_g: Box<dyn Regressor>,
55 ml_pi: Box<dyn Classifier>,
56 ml_m: Box<dyn Classifier>,
57 score: SsmScore,
58 clipping_threshold: f64,
59 normalize_ipw: bool,
60 splitting: SampleSplitting,
61}
62
63impl Ssm {
64 pub fn new(
67 ml_g: impl Regressor + 'static,
68 ml_pi: impl Classifier + 'static,
69 ml_m: impl Classifier + 'static,
70 ) -> Self {
71 Self::from_boxed(Box::new(ml_g), Box::new(ml_pi), Box::new(ml_m))
72 }
73
74 pub fn from_boxed(
75 ml_g: Box<dyn Regressor>,
76 ml_pi: Box<dyn Classifier>,
77 ml_m: Box<dyn Classifier>,
78 ) -> Self {
79 Self {
80 ml_g,
81 ml_pi,
82 ml_m,
83 score: SsmScore::default(),
84 clipping_threshold: 1e-2,
85 normalize_ipw: false,
86 splitting: SampleSplitting::default(),
87 }
88 }
89
90 pub fn score(mut self, score: SsmScore) -> Self {
91 self.score = score;
92 self
93 }
94
95 pub fn clipping_threshold(mut self, t: f64) -> Self {
98 self.clipping_threshold = t;
99 self
100 }
101
102 pub fn normalize_ipw(mut self, normalize: bool) -> Self {
104 self.normalize_ipw = normalize;
105 self
106 }
107
108 dml_model_api!();
109
110 fn nonignorable_nuisances(
113 &self,
114 input: &ScoreInput<'_>,
115 s: ArrayView1<'_, f64>,
116 z: ArrayView1<'_, f64>,
117 folds: &[Fold],
118 ) -> Result<[Array1<f64>; 4]> {
119 let ScoreInput { y, d, x, .. } = *input;
120 let n = y.len();
121 let strata: Vec<f64> = (0..n).map(|i| d[i] + 2.0 * s[i]).collect();
122 let dxz = concatenate![Axis(1), x, d.insert_axis(Axis(1)), z.insert_axis(Axis(1))];
123 let fit_fold = |fold: &Fold| -> Result<[Array1<f64>; 4]> {
124 let (train1, train2) = half_split(&fold.train, &strata)?;
125 let pi_all = fit_proba(self.ml_pi.as_ref(), dxz.view(), &train1, s, dxz.view())?;
128 let xpi = concatenate![Axis(1), x, pi_all.view().insert_axis(Axis(1))];
129 let xpi_test = xpi.select(Axis(0), &fold.test);
130 let m = fit_proba(self.ml_m.as_ref(), xpi.view(), &train2, d, xpi_test.view())?;
131 let mut sorted2 = train2.clone();
134 sorted2.sort_unstable();
135 let outcome = |level: f64| -> Result<Array1<f64>> {
136 let rows: Vec<usize> = sorted2
137 .iter()
138 .copied()
139 .filter(|&i| d[i] == level && s[i] == 1.0)
140 .collect();
141 if rows.is_empty() {
142 return Err(DmlError::InvalidArgument(
143 "a training fold has no selected observations in a treatment arm".into(),
144 ));
145 }
146 let mut model = self.ml_g.fresh();
147 model.fit(xpi.select(Axis(0), &rows).view(), gather(y, &rows).view())?;
148 model.predict(xpi_test.view())
149 };
150 let pi_test = gather(pi_all.view(), &fold.test);
151 Ok([outcome(0.0)?, outcome(1.0)?, pi_test, m])
152 };
153 #[cfg(feature = "parallel")]
154 let per_fold: Vec<[Array1<f64>; 4]> = {
155 use rayon::prelude::*;
156 folds.par_iter().map(fit_fold).collect::<Result<_>>()?
157 };
158 #[cfg(not(feature = "parallel"))]
159 let per_fold: Vec<[Array1<f64>; 4]> = folds.iter().map(fit_fold).collect::<Result<_>>()?;
160
161 let mut out: [Array1<f64>; 4] = std::array::from_fn(|_| Array1::from_elem(n, f64::NAN));
162 for (fold, preds) in folds.iter().zip(per_fold) {
163 for (dst, src) in out.iter_mut().zip(preds) {
164 for (k, &i) in fold.test.iter().enumerate() {
165 dst[i] = src[k];
166 }
167 }
168 }
169 Ok(out)
170 }
171}
172
173fn fit_proba(
175 learner: &dyn Classifier,
176 x: ArrayView2<'_, f64>,
177 rows: &[usize],
178 labels: ArrayView1<'_, f64>,
179 x_pred: ArrayView2<'_, f64>,
180) -> Result<Array1<f64>> {
181 let mut model = learner.fresh();
182 model.fit(x.select(Axis(0), rows).view(), gather(labels, rows).view())?;
183 model.predict_proba(x_pred)
184}
185
186impl LinearScoreModel for Ssm {
187 fn name(&self) -> String {
188 match self.score {
189 SsmScore::MissingAtRandom => "SSM (missing at random)".into(),
190 SsmScore::Nonignorable => "SSM (nonignorable)".into(),
191 }
192 }
193
194 fn validate(&self, data: &DmlData) -> Result<()> {
195 if !(0..data.n_treat()).all(|j| data.is_binary_treatment(j)) {
196 return Err(DmlError::InvalidArgument(
197 "SSM needs binary (0/1) treatments".into(),
198 ));
199 }
200 if data.selection().is_none() {
201 return Err(DmlError::InvalidArgument(
202 "SSM needs a selection indicator (DmlData::with_selection_indicator)".into(),
203 ));
204 }
205 if self.score == SsmScore::Nonignorable && data.z().is_none() {
206 return Err(DmlError::InvalidArgument(
207 "the nonignorable score needs an instrument for selection (DmlData::with_instrument)"
208 .into(),
209 ));
210 }
211 if !(self.clipping_threshold > 0.0 && self.clipping_threshold < 0.5) {
212 return Err(DmlError::InvalidArgument(
213 "clipping_threshold must be in (0, 0.5)".into(),
214 ));
215 }
216 Ok(())
217 }
218
219 fn strata(&self, data: &DmlData) -> Option<Vec<usize>> {
221 let (d, s) = (data.treatment(0), data.selection()?);
222 Some(
223 d.iter()
224 .zip(s)
225 .map(|(&d, &s)| (d + 2.0 * s) as usize)
226 .collect(),
227 )
228 }
229
230 fn estimate_score(&self, input: &ScoreInput<'_>, folds: &[Fold]) -> Result<ScoreEstimate> {
231 let ScoreInput { y, d, x, z, s, .. } = *input;
232 let s =
233 s.ok_or_else(|| DmlError::InvalidArgument("SSM needs a selection indicator".into()))?;
234 if !is_binary(d) {
235 return Err(DmlError::InvalidArgument(
236 "SSM needs a binary treatment".into(),
237 ));
238 }
239 let n = y.len();
240 let [g0, g1, pi, m] = match self.score {
241 SsmScore::MissingAtRandom => {
242 let dx = concatenate![Axis(1), x, d.insert_axis(Axis(1))];
243 let pi = cross_fit_predict_proba(self.ml_pi.as_ref(), dx.view(), s, folds)?;
244 let m = cross_fit_predict_proba(self.ml_m.as_ref(), x, d, folds)?;
245 let selected = |level: f64| -> Vec<bool> {
246 (0..n).map(|i| d[i] == level && s[i] == 1.0).collect()
247 };
248 let g1 = cross_fit_predict_subset(self.ml_g.as_ref(), x, y, folds, &selected(1.0))?;
249 let g0 = cross_fit_predict_subset(self.ml_g.as_ref(), x, y, folds, &selected(0.0))?;
250 [g0, g1, pi, m]
251 }
252 SsmScore::Nonignorable => {
253 let z = z.ok_or_else(|| {
254 DmlError::InvalidArgument("the nonignorable score needs an instrument".into())
255 })?;
256 self.nonignorable_nuisances(input, s, z, folds)?
257 }
258 };
259 let t = self.clipping_threshold;
260 let m = m.mapv(|p| p.clamp(t, 1.0 - t));
261
262 let treat: Array1<f64> = d.mapv(|v| f64::from(v == 1.0));
266 let control: Array1<f64> = d.mapv(|v| f64::from(v == 0.0));
267 let (w1, w0) = if self.normalize_ipw {
268 let sum = |a: Array1<f64>| a.iter().fold(0.0, |acc, &v| acc + v);
269 let w1 = treat.sum() / sum(&treat * &s / &(&pi * &m));
270 let w0 = control.sum() / sum(&control * &s / &(&pi * &(1.0 - &m)));
271 (Some(w1), Some(w0))
272 } else {
273 (None, None)
274 };
275 let psi_b: Array1<f64> = (0..n)
276 .map(|i| {
277 let c1 = treat[i] * s[i] * (y[i] - g1[i]) / (m[i] * pi[i]);
278 let c0 = control[i] * s[i] * (y[i] - g0[i]) / ((1.0 - m[i]) * pi[i]);
279 let b1 = w1.map_or(c1, |w| w * c1) + g1[i];
280 let b0 = w0.map_or(c0, |w| w * c0) + g0[i];
281 b1 - b0
282 })
283 .collect();
284 let psi_a = Array1::from_elem(n, -1.0);
285 Ok(ScoreEstimate::new(
286 LinearScore::new(psi_a, psi_b)?,
287 vec![("ml_g_d0", g0), ("ml_g_d1", g1), ("ml_pi", pi), ("ml_m", m)],
288 )
289 .with_targets(vec![
291 NuisanceTarget::regression("ml_g_d0", y.to_owned()),
292 NuisanceTarget::regression("ml_g_d1", y.to_owned()),
293 NuisanceTarget::classification("ml_pi", s.to_owned()),
294 NuisanceTarget::classification("ml_m", d.to_owned()),
295 ]))
296 }
297}