antecedent.model

Interventional sampling helpers.

 1"""Interventional sampling helpers."""
 2
 3from __future__ import annotations
 4
 5from ._native import (
 6    FittedGcm,
 7    GcmSampleResult,
 8    fit_gcm,
 9    sample_do,
10    sample_interventional_distribution,
11)
12
13__all__ = [
14    "FittedGcm",
15    "GcmSampleResult",
16    "fit_gcm",
17    "sample_do",
18    "sample_interventional_distribution",
19]
class FittedGcm:

Fitted GCM retained across Python calls (fit once, sample / attribute many times).

Model, data, and names are [Arc]-shared so GIL-detached methods clone handles instead of deep-copying the fitted session on every click.

def mechanism_kinds(self, /):

Selected mechanism family id per variable name.

def sample_do(self, /, interventions, n, *, seed=0, threads=1):

Interventional ancestral sample under hard do values.

def counterfactual_ite(self, /, treatment, outcome, active, control, *, seed=0, threads=1):

Unit-level ITE under hard interventions on treatment.

def attribute_path_specific( self, /, treatment, outcome, *, path_nodes=None, max_paths=64, max_len=16, seed=0, threads=1):
def attribute_paths( self, /, sources, outcome, *, max_paths=64, max_len=16, seed=0, threads=1):
def attribute_distribution_change( self, /, outcome, baseline_start, baseline_end, comparison_start, comparison_end, *, n_samples=500, seed=0, threads=1):
def attribute_distribution_change_robust( self, /, outcome, baseline_start, baseline_end, comparison_start, comparison_end, *, n_samples=500, seed=0, threads=1):
def attribute_structure_change( self, /, comparison_edges, outcome, baseline_start, baseline_end, comparison_start, comparison_end, *, n_samples=500, seed=0, threads=1):
def attribute_unit_change(self, /, outcome, *, max_units=0, seed=0, threads=1):
def attribute_feature_relevance(self, /, outcome, *, delta=1.0, n_samples=200, seed=0, threads=1):
def anomaly_attribution(self, /, outcomes, *, max_units=0):
def mechanism_change_detection( self, /, baseline_start, baseline_end, comparison_start, comparison_end, *, seed=0, threads=1):
def rank_root_causes(self, /, attribution, *, seed=0, threads=1):
n_assignments
names
class GcmSampleResult:

Interventional samples under hard do (means + full draws).

column_means
n_nodes
n_draws
draws

Column-major draws shaped (n_nodes, n_draws).

def fit_gcm(names, columns, edges, *, threads=1):

Fit a linear-Gaussian GCM; return a reusable [FittedGcm].

def sample_do( names, columns, edges, treatment, do_value, n_draws, *, seed=0, threads=1, mechanism_wrappers=None):

Fit GCM and return interventional column means + draws under hard do(treatment=value).

mechanism_wrappers maps variable name → object with sample_noise(n) / evaluate(parents, noise) ( slow path).

def sample_interventional_distribution( names, columns, edges, treatment, do_value, n_draws, outcome=None, *, seed=0, threads=1):

Sample an interventional distribution via [InterventionalDistributionQuery].

Same return shape as [gcm_sample_do]; builds the typed query then samples.