causasv 0.8.9

Causal feature attribution via Asymmetric Shapley Values (ASV) over DAGs — Rust + Python
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98

"""Tests for ASVEnsembleExplainer."""

import pytest

from causasv import ASVEnsembleExplainer, ASVExplainer, CausalDAG


def additive_fn(names):
    return float(len(names))


@pytest.fixture
def chain_dag():
    return CausalDAG.from_edges([("a", "b"), ("b", "c")])


@pytest.fixture
def diamond_dag():
    return CausalDAG.from_edges([("a", "b"), ("a", "c"), ("b", "d"), ("c", "d")])


def test_ensemble_single_dag_matches_explainer(chain_dag):
    """Ensemble of one DAG must return the same values as a direct ASVExplainer."""
    single = ASVExplainer(chain_dag).explain(additive_fn, method="exact")
    result = ASVEnsembleExplainer([chain_dag]).explain_with_sensitivity(
        additive_fn, method="exact"
    )
    for k, v in single.items():
        assert abs(result["mean_values"][k] - v) < 1e-9


def test_ensemble_single_dag_std_is_zero(chain_dag):
    """Single DAG → std = 0 for all features."""
    result = ASVEnsembleExplainer([chain_dag]).explain_with_sensitivity(
        additive_fn, method="exact"
    )
    for v in result["std_values"].values():
        assert abs(v) < 1e-12


def test_ensemble_rank_stability_identical_dags():
    """Identical DAGs with a non-symmetric value fn → rank_stability = 1.0."""
    dag = CausalDAG.from_edges([("a", "b"), ("b", "c")])

    def nonlinear(names):
        s = set(names)
        return ("a" in s) * 3 + ("b" in s) * 2 + ("c" in s) * 6 * ("a" in s and "c" in s)

    result = ASVEnsembleExplainer([dag, dag]).explain_with_sensitivity(nonlinear, method="exact")
    assert abs(result["rank_stability"] - 1.0) < 1e-9


def test_ensemble_result_keys(chain_dag, diamond_dag):
    result = ASVEnsembleExplainer([chain_dag]).explain_with_sensitivity(additive_fn, method="exact")
    assert set(result.keys()) == {"mean_values", "std_values", "rank_stability", "per_dag_values"}


def test_ensemble_per_dag_values_count():
    dag1 = CausalDAG.from_edges([("a", "b"), ("b", "c")])
    dag2 = CausalDAG.from_edges([("a", "b"), ("a", "c")])
    dag3 = CausalDAG.from_edges([("a", "c"), ("b", "c")])
    result = ASVEnsembleExplainer([dag1, dag2, dag3]).explain_with_sensitivity(
        additive_fn, method="exact"
    )
    assert len(result["per_dag_values"]) == 3


def test_ensemble_rank_stability_range():
    """rank_stability must be in [-1, 1]."""
    dag1 = CausalDAG.from_edges([("a", "b"), ("b", "c")])
    dag2 = CausalDAG.from_edges([("a", "b"), ("a", "c")])
    result = ASVEnsembleExplainer([dag1, dag2]).explain_with_sensitivity(
        additive_fn, method="exact"
    )
    assert -1.0 <= result["rank_stability"] <= 1.0


def test_ensemble_mean_equals_average():
    """mean_values must equal the arithmetic mean of per_dag_values."""
    dag1 = CausalDAG.from_edges([("a", "b"), ("b", "c")])
    dag2 = CausalDAG.from_edges([("a", "b"), ("a", "c")])
    result = ASVEnsembleExplainer([dag1, dag2]).explain_with_sensitivity(
        additive_fn, method="exact"
    )
    for f in result["mean_values"]:
        expected = sum(d.get(f, 0.0) for d in result["per_dag_values"]) / len(
            result["per_dag_values"]
        )
        assert abs(result["mean_values"][f] - expected) < 1e-9


def test_ensemble_empty():
    """Empty ensemble returns empty dicts and rank_stability=1.0."""
    result = ASVEnsembleExplainer([]).explain_with_sensitivity(additive_fn)
    assert result["mean_values"] == {}
    assert result["std_values"] == {}
    assert result["rank_stability"] == 1.0
    assert result["per_dag_values"] == []