causalig.utils

Graph utilities

All functions take an adjacency_matrix with adjacency_matrix[i, j] != 0 meaning i -> j.

causalig.utils.find_ancestors(causal_nodes, adjacency_matrix)

Compute all ancestor nodes (direct AND indirect parents) of a set of nodes in a directed acyclic graph.

Parameters:

• causal_nodes (iterable of ints) – The nodes whose ancestors we want.

• adjacency_matrix (2D array-like (p x p)) – Adjacency matrix where adjacency_matrix[i, j] != 0 means i -> j.

Returns:

ancestors (set of ints) – All ancestors of the causal nodes.

causalig.utils.find_eldest_ancestors(causal_nodes, adjacency_matrix)

Find the eldest ancestors among the causal nodes. An eldest ancestor is a causal node that has no parents within the set of causal nodes.

Parameters:

• causal_nodes (iterable of ints) – The nodes considered as causal.

• adjacency_matrix (2D array-like (p x p)) – Adjacency matrix where adjacency_matrix[i, j] != 0 means i -> j.

Returns:

eldest_ancestors (set of ints) – The eldest ancestors among the causal nodes.

causalig.utils.graph.get_connected_groups(adjacency_matrix)

BFS on a directed graph (adjacency matrix, rows=sources, cols=targets). Returns a list of groups, where each group is a list of node indices reachable from a root (in-degree 0, out-degree > 0). Isolated or cyclic nodes are appended as separate singleton groups.

causalig.utils.graph.propagate_upstream(attributions, adjacency_matrix, n_hops=1, push_fraction=1.0)

Redistribute feature attributions upstream along a causal DAG.

Model- and method-agnostic post-processing step: takes attributions already produced by any attribution method (IG, SHAP, …) and pushes them toward upstream ancestors using only the graph topology, with no knowledge of the model or data. Each round, every feature sends push_fraction of its currently held mass to its parents (split evenly across them) and keeps the rest; a feature with no parents (a root) has nowhere to send to, so it keeps its mass regardless of push_fraction. Total attribution mass is conserved exactly across rounds.

Parameters:

• attributions ((d,) or (batch, d) tensor) – Per-feature attributions from any attribution method.

• adjacency_matrix ((d, d) tensor/array) – Adjacency matrix where adjacency_matrix[i, j] != 0 means i -> j.

• n_hops (int) – Number of propagation rounds. n_hops=0 returns attributions unchanged.

• push_fraction (float) – Fraction, in [0, 1], of a node’s mass sent to its parents each round. The rest is retained at that node.

Returns:

Tensor of the same shape as attributions.

Return type:

Tensor

Logging

causalig.utils.setup_experiment_logger(experiment_name, log_dir='logs', log_level=20, console_output=True)

Set up a logger for an experiment.

Parameters:

• experiment_name (str) – Name of the experiment (used in log filename)

• log_dir (str) – Directory to save log files (default: “logs”)

• log_level (int) – Logging level (default: logging.INFO)

• console_output (bool) – Whether to also output to console (default: True)

Returns:

Configured logger instance

Return type:

Logger

causalig.utils.get_experiment_logger(experiment_name=None)

Get an existing logger or create a default one.

Parameters:

experiment_name (str | None) – Name of the experiment. If None, uses default name.

Returns:

Logger instance

Return type:

Logger