py-bbn

Common

class pybbn.common.GraphTuple(u: Graph, d: DiGraph)

Bases: object

Graph tuple of DAG and its undirected version.

__init__(u: Graph, d: DiGraph) → None

Ctor.

Parameters:

u – Undirected graph.
d – DAG.

property d: DiGraph

Get the DAG.

Returns:: DAG.

property u: Graph

Get the undirected graph.

Returns:: Undirected graph.

class pybbn.common.IReasoningModel

Bases: ABC

Reasoning model abstract base class.

abstract property assignment: dict[Any, Any]

Node to cluster assignments.

Returns:: Node to cluster assignments.

abstract property c: list[set[Any]]

Clusters induced by triangulation and used for join tree.

Returns:: Clusters.

abstract property cluster_potentials: dict[Any, DataFrame]

Cluster potentials.

Returns:: Cluster potentials.

abstract cquery(Y: str, e: dict[str, str], h: dict[str, str]) → DataFrame

Counterfactual query.

Computes the counterfactual: Given the evidence, \(e\), what would \(P(Y_h)\) be had the hypothetical \(h\) happened? The evidence is the factual. The hypothetical is the counterfactual.

Parameters:

Y – Target variable of interest.
e – Evidence (factual).
h – Hypothetical (counterfactual).

Returns:

Counterfactual estimation of Y.

abstract create_finding_evidences(n: Any, evidences: list[float], values: list[str] | None = None) → DataFrame

Create finding evidences. The values should be either 1 or 0. The order of the evidences should correspond to the sorted values of the node/variable when no values are specified.

Parameters:

n – Node.
evidences – Evidences.
values – The values of the node/variable corresponding to the evidences.

Returns:

Potential.

abstract create_observation_evidences(n: Any, v: str) → DataFrame

Create observation evidences. Only one value is set to 1 and the others are set to 0.

Parameters:

n – Node.
v – Value that is observed.

Returns:

Potential.

abstract property d: DiGraph

DAG.

Returns:: DAG.

abstract property domains: dict[Any, list[Any]]

Get the domains.

Returns:: Domains.

abstract e(evidences: dict[Any, str]) → dict[Any, DataFrame]

Shortcut to create observation evidences.

Parameters:: evidences – Observation evidences.
Returns:: Evidences.

abstract get_observation_evidences(evidences: dict[Any, str]) → dict[Any, DataFrame]

Create observation evidences via key-value pairs. Easier to work with.

Parameters:: evidences – Observation evidences.
Returns:: Evidences.

abstract property graph_tuple: GraphTuple

Graph tuple.

Returns:: Graph tuple.

abstract iquery(Y: list[Any], y: list[str], X: list[Any], x: list[str], x_ref: list[str] | None = None) → Series

Interventional query.

Perform the do-operation using the (backdoor) adjustment formula.

\(P(Y=y|do(X=x)) = \sum_{z \in Z} P(Y=y|X=x, Z=z)P(Z=z)\)

The output is the causal effect. To estimate the average causal effect, either call this function twice or provide the optional x_ref argument to compute the following quantity in a single call.

\(P(Y=y|do(X=x_1)) - P(Y=y|do(X=x_2))\)

Parameters:

Y – Set of variables Y.
y – Set of values corresponding to Y.
X – Set of variables X.
x – Set of values corresponding to X.
x_ref – Optional reference intervention values for X. When provided, the returned series is the difference \(P(Y=y|do(X=x)) - P(Y=y|do(X=x' ))\) where \(x'\) corresponds to x_ref. Each element of x_ref must align with the variables in X.

Returns:

\(P(Y=y|do(X=x))\) or the difference with the reference intervention when x_ref is provided.

abstract property j: Graph

Join tree graph.

Returns:: Join tree graph.

abstract property m: Graph

Moralized graph.

Returns:: Moralized graph.

abstract property messages: list[tuple[Any, Any, Any]]

List of message passes.

Returns:: Message passes.

abstract property node_potentials: dict[Any, DataFrame]

Node potentials.

Returns:: Node potentials.

abstract pquery(nodes: list[Any] | None = None, evidences: dict[Any, DataFrame] | None = None) → dict[Any, DataFrame]

Query the model.

Parameters:

nodes – Nodes that we want returned. If not provided then all nodes are returned.
evidences – Evidences.

Returns:

Potentials.

abstract sample(max_samples: int = 100, evidence: dict[Any, str] | None = None, seed: int = 37) → DataFrame

Applies logic sampling to generate samples.

Parameters:

max_samples – Maximum number of samples.
evidence – Any evidence.
seed – Seed. Default is 37.

Returns:

Samples.

abstract sample_twin(max_samples: int = 100, seed: int = 37) → DataFrame

Samples data for twin network.

Parameters:

max_samples – Maximum number of samples.
seed – Seed. Default is 37.

Returns:

Samples.

abstract property t: Graph

Triangulated graph.

Returns:: Triangulated graph.

abstract property u: Graph

Undirected graph of the DAG.

Returns:: Undirected graph.

Factory

pybbn.factory.create_reasoning_model(d: DiGraph | dict[str, Any], p: dict[Any, list[dict[str, Any]] | dict[str, Any]]) → IReasoningModel

Create reasoning model.

Parameters:

d – DAG. Could be a nx.DiGraph or a dictionary.
p – Parameters (CPTs).

Returns:

Reasoning model.

Parameter

pybbn.parameter.get_conditional_cpt(pa: list[Any], ch: Any, df: DataFrame) → dict[str, Any]

Get conditional CPT.

Parameters:

pa – Parents.
ch – Child.
df – Data.

Returns:

Conditional probability table.

pybbn.parameter.get_marginal_cpt(f: Any, df: DataFrame) → dict[str, Any]

Learn marginal CPT.

Parameters:

f – Node.
df – Data.

Returns:

Marginal probability table.

pybbn.parameter.learn_parameters(d: DiGraph, df: DataFrame) → dict[str, dict[str, Any]]

Learn the parameters of each node in the DAG.

Parameters:

d – DAG.
df – Data.

Returns:

Parameters.

Reasoning

class pybbn.reasoning.ReasoningModel(d: DiGraph, u: Graph, m: Graph, t: Graph, j: Graph, c: list[set[Any]], domains: dict[Any, list[Any]], messages: list[tuple[Any, Any, Any]], assignment: dict[Any, Any], node_potentials: dict[Any, DataFrame], cluster_potentials: dict[Any, DataFrame])

Bases: IReasoningModel

Reasoning model.

__init__(d: DiGraph, u: Graph, m: Graph, t: Graph, j: Graph, c: list[set[Any]], domains: dict[Any, list[Any]], messages: list[tuple[Any, Any, Any]], assignment: dict[Any, Any], node_potentials: dict[Any, DataFrame], cluster_potentials: dict[Any, DataFrame]) → None

Ctor.

Parameters:

d – DAG.
u – Undirected graph of the DAG.
m – Moralized graph.
t – Triangulated graph.
j – Join tree.
c – Clusters.
messages – Messages.
assignment – Assignment mapping of nodes to clusters.
node_potentials – Node potentials.
cluster_potentials – Cluster potentials.

property assignment: dict[Any, Any]

Node to cluster assignments.

Returns:: Node to cluster assignments.

property c: list[set[Any]]

Clusters induced by triangulation and used for join tree.

Returns:: Clusters.

property cluster_potentials: dict[Any, DataFrame]

Cluster potentials.

Returns:: Cluster potentials.

cquery(Y: str, e: dict[str, str], h: dict[str, str]) → DataFrame

Counterfactual query.

Computes the counterfactual: Given the evidence, \(e\), what would \(P(Y_h)\) be had the hypothetical \(h\) happened? The evidence is the factual. The hypothetical is the counterfactual.

Parameters:

Y – Target variable of interest.
e – Evidence (factual).
h – Hypothetical (counterfactual).

Returns:

Counterfactual estimation of Y.

create_finding_evidences(n: Any, evidences: list[float], values: list[str] | None = None) → DataFrame

Create finding evidences. The values should be either 1 or 0. The order of the evidences should correspond to the sorted values of the node/variable when no values are specified.

Parameters:

n – Node.
evidences – Evidences.
values – The values of the node/variable corresponding to the evidences.

Returns:

Potential.

create_observation_evidences(n: Any, v: str) → DataFrame

Create observation evidences. Only one value is set to 1 and the others are set to 0.

Parameters:

n – Node.
v – Value that is observed.

Returns:

Potential.

create_virtual_evidences(n: Any, evidences: list[float], values: list[str] | None = None) → DataFrame

Create virtual evidence. The values should be in the range [0, 1]. The order of the evidences should correspond to the sorted values of the node/variable.

Parameters:

n – Node.
evidences – Evidences.
values – The values of the node/variable corresponding to the evidences.

Returns:

Potential.

property d: DiGraph

DAG.

Returns:: DAG.

property domains: dict[Any, list[Any]]

Get the domains.

Returns:: Domains.

e(evidences: dict[Any, str]) → dict[Any, DataFrame]

Shortcut to create observation evidences.

Parameters:: evidences – Observation evidences.
Returns:: Evidences.

get_all_confounders(x: Any, y: Any) → list[Any]

Get the all confounders wrt to x and y.

Parameters:

x – Node.
y – Node.

Returns:

List of all confounders.

get_all_mediators(x: Any, y: Any) → list[Any]

Get the all mediators wrt to x and y.

Parameters:

x – Node.
y – Node

Returns:

List of all mediators.

get_all_paths(x: Any, y: Any) → list[tuple[Any]]

Get all paths from x to y.

Parameters:

x – Start node.
y – Stop node.

Returns:

All simple paths between start and stop nodes.

get_minimal_confounders(x: Any, y: Any) → list[Any]

Get the minimal confounders wrt to x and y.

Parameters:

x – Node.
y – Node.

Returns:

List of minimal confounders.

get_minimal_mediators(x: Any, y: Any) → list[Any]

Get the minimal mediators wrt to x and y.

Parameters:

x – Node.
y – Node.

Returns:

List of minimal mediators.

get_observation_evidences(evidences: dict[Any, str]) → dict[Any, DataFrame]

Create observation evidences via key-value pairs. Easier to work with.

Parameters:: evidences – Observation evidences.
Returns:: Evidences.

property graph_tuple: GraphTuple

Graph tuple.

Returns:: Graph tuple.

iquery(Y: list[Any], y: list[str], X: list[Any], x: list[str], method: str = 'backdoor', x_ref: list[str] | None = None) → Series

Interventional query.

Perform the do-operation using the (backdoor) adjustment formula.

\(P(Y=y|do(X=x)) = \sum_{z \in Z} P(Y=y|X=x, Z=z)P(Z=z)\)

The output is the causal effect. To estimate the average causal effect, either call this function twice or provide the optional x_ref argument to compute the following quantity.

\(P(Y=y|do(X=x_1)) - P(Y=y|do(X=x_2))\)

Parameters:

Y – Set of variables Y.
y – Set of values corresponding to Y.
X – Set of variables X.
x – Set of values corresponding to X.
method – Method used to perform the do-operation. Must be one of ‘backdoor’ or ‘graphical’.
x_ref – Optional reference intervention values for X. When provided, the returned series equals \(P(Y=y|do(X=x)) - P(Y=y|do(X=x'))\), where \(x'\) corresponds to x_ref.

Returns:

\(P(Y=y|do(X=x))\) or the difference with the reference intervention when x_ref is provided.

is_active_path(path: list[Any], evidence: set[Any] | None = None) → bool

Determine if a path is active given a set of evidence nodes.

Parameters:

path – Path as a list of nodes.
evidence – Set of evidence nodes.

Returns:

True if the path is active, False otherwise.

is_d_separated(x: Any, y: Any, evidence: set[Any] | None = None) → bool

Determine if x and y are d-separated.

Parameters:

x – Node.
y – Node.
evidence – Set of evidence nodes.

Returns:

True if of x and y are d-separated, False otherwise.

property j: Graph

Join tree graph.

Returns:: Join tree graph.

property m: Graph

Moralized graph.

Returns:: Moralized graph.

property messages: list[tuple[Any, Any, Any]]

List of message passes.

Returns:: Message passes.

property node_potentials: dict[Any, DataFrame]

Node potentials.

Returns:: Node potentials.

pquery(nodes: list[Any] | None = None, evidences: dict[Any, DataFrame] | None = None) → dict[Any, DataFrame]

Probabilistic query.

Parameters:

nodes – Nodes that we want returned. If not provided then all nodes are returned.
evidences – Evidences.

Returns:

Potentials.

sample(max_samples: int = 100, evidence: dict[Any, str] | None = None, seed: int = 37) → DataFrame

Applies logic sampling to generate samples.

Parameters:

max_samples – Maximum number of samples.
evidence – Any evidence.
seed – Seed. Default is 37.

Returns:

Samples.

sample_twin(max_samples: int = 100, seed: int = 37) → DataFrame

Samples data for twin network.

Parameters:

max_samples – Maximum number of samples.
seed – Seed. Default is 37.

Returns:

Samples.

scm_equation(node: Any) → IReasoningModel

Returns SCM equation corresponding to node.

Returns:: SCMs.

property t: Graph

Triangulated graph.

Returns:: Triangulated graph.

property u: Graph

Undirected graph of the DAG.

Returns:: Undirected graph.

Associational Query

class pybbn.associational.ArrayPotential(values: ndarray, axes: list[str])

Bases: object

Lightweight numpy representation of a potential.

__init__(values: ndarray, axes: list[str]) → None

axes: list[str]

values: ndarray

pybbn.associational.array_to_potential(p: ArrayPotential, domains: dict[Any, list[Any]]) → DataFrame: Convert an ArrayPotential back to a DataFrame potential.

pybbn.associational.clone_array_potential_map(m: dict[Any, ArrayPotential]) → dict[Any, ArrayPotential]: Clone a mapping of array potentials.

pybbn.associational.clone_potential_map(m: dict[Any, DataFrame]) → dict[Any, DataFrame]

Clones the potentials.

Parameters:: m – Potentials.
Returns:: Potentials.

pybbn.associational.cpt_to_pot(node: Any, cpts: dict[Any, list[dict[str, Any]] | dict[str, Any]]) → DataFrame

Create potential of the specified node.

Parameters:

node – Node.
cpts – CPTs.

Returns:

Potential.

pybbn.associational.create_finding_evidences(n: Any, values: list[str], evidences: list[float]) → DataFrame

Create finding evidences. Finding evidence values must be either 0 or 1.

Parameters:

n – Node.
values – Domain (list of values).
evidences – Evidences.

Returns:

Potential

pybbn.associational.create_observation_evidences(n: Any, values: list[str], v: str) → DataFrame

Create observation evidences. Observation evidence must have only one 1 and the rest 0’s.

Parameters:

n – Node.
values – Domain (list of values).
v – The value that is observed.

Returns:

Potential

pybbn.associational.create_sepsets(g: Graph, domains: dict[Any, list[Any]]) → DataFrame

Creates all possible separation sets.

Parameters:

g – Initialized join tree.
domains – Domains.

Returns:

Data frame of information on separation sets that can be created. Sorted by mass (desc) and cost (asc).

pybbn.associational.create_virtual_evidences(n: Any, values: list[str], evidences: list[float]) → DataFrame

Create virtual evidences. Virtual evidence values must be in the range of [0, 1].

Parameters:

n – Node.
values – Domain (list of values).
evidences – Evidences.

Returns:

Potential

pybbn.associational.cwalk(x: Any, s: Any, y: Any, g: Graph, marked: list[Any], messages: list[tuple[Any, Any, Any]]) → None

Recursively walk the graph to get the collect message passes.

Parameters:

x – X node (cluster/clique).
s – S node (sepset).
y – Y node (cluster/clique).
g – Join tree.
marked – List of marked clusters.
messages – List of message passes.

pybbn.associational.dict_to_graph(d: dict) → DiGraph

Convert a dictionary specifying a DAG to a networkx DAG.

Parameters:: d – A dictionary.
Returns:: A networkx DAG.

pybbn.associational.divide(N: DataFrame, D: DataFrame, columns: list[str] | None = None) → DataFrame

Divide two potentials N / D where N is the numerator and D the denominator.

Parameters:

N – Potential.
D – Potential.
columns – Precomputed common nodes between N and D. If None then the common nodes are computed on the fly.

Returns:

Divided potential.

pybbn.associational.divide_np(N: ArrayPotential, D: ArrayPotential) → ArrayPotential: Divide two array potentials using numpy.divide.

pybbn.associational.dwalk(x: Any, g: Graph, marked: list[Any], messages: list[tuple[Any, Any, Any]]) → None

Recursively walk the graph to get the distribute message passes.

Parameters:

x – X node (cluster/clique).
g – Join tree.
marked – List of marked clusters.
messages – List of message passes.

pybbn.associational.get_all_triangulation_info(g: Graph, domains: dict[Any, list[Any]]) → DataFrame

Get all the triangulation information for each node in the graph.

Parameters:

g – Graph (moralized).
domains – Domains.

Returns:

Triangulation information (e.g. weight and edges). Already sorted by edges/weight ascendingly.

pybbn.associational.get_best_triangulation(g: Graph, domains: dict[Any, list[Any]]) → Any

Get the node that is the best one for triangulation.

Parameters:

g – Graph (moralized).
domains – Domains.

Returns:

Node.

pybbn.associational.get_cluster_id(cluster: set[Any]) → str

Create cluster id.

Parameters:: cluster – Cluster.
Returns:: Cluster id.

pybbn.associational.get_cluster_weight(nodes: list[Any], domains: dict[Any, list[Any]]) → int

Get the cluster weight.

Parameters:

nodes – Nodes.
domains – Domains.

Returns:

Weight.

pybbn.associational.get_collect_messages(x: Any, g: Graph) → list[tuple[Any, Any, Any]]

Get the collect evidence messages.

Parameters:

x – X node.
g – Join tree.

Returns:

List of message pass.

pybbn.associational.get_common_axes(X: ArrayPotential, Y: ArrayPotential) → list[str]: Get common axes between two array potentials.

pybbn.associational.get_common_nodes(X: DataFrame, Y: DataFrame) → list[str]

Get the common nodes between two potentials.

Parameters:

X – Potential.
Y – Potential.

Returns:

List of common nodes.

pybbn.associational.get_distribute_messages(x: Any, g: Graph) → list[tuple[Any, Any, Any]]

Get the distribute evidence messages.

Parameters:

x – X node.
g – Join tree.

Returns:

List of message pass.

pybbn.associational.get_domains(p_map: dict[Any, DataFrame]) → dict[Any, list[Any]]

Get the domains of each variable/node from the potentials.

Parameters:: p_map – Potential map.
Returns:: Domain map.

pybbn.associational.get_join_tree(clusters: list[set[Any]], domains: dict[Any, list[Any]]) → Graph

Get a join tree.

Parameters:

clusters – Clusters.
domains – Domains.

Returns:

Join tree.

pybbn.associational.get_messages(x: Any, g: Graph) → list[tuple[Any, Any, Any]]

Get the collect and distribute message passes.

Parameters:

x – X node.
g – Join tree.

Returns:

List of message passes.

pybbn.associational.get_node_to_cluster_assignment(d: DiGraph, j: Graph) → dict[Any, Any]

Assign each node to a cluster. The assignment is such that the cluster has the node and its parents.

Parameters:

d – DAG.
j – Join tree.

Returns:

Node to cluster assignments.

pybbn.associational.get_node_weight(node: Any, domains: dict[Any, list[Any]]) → int

Get the node weight.

Parameters:

node – Node.
domains – Domains.

Returns:

Weight.

pybbn.associational.get_num_required_edges(g: Graph, node: Any) → int

Get the number of required edges that will be created by connecting all the neighbors of node. This avoids enumerating edge pairs and instead counts them using set intersections.

Parameters:

g – Graph (moralized graph).
node – Node.

Returns:

Number of required edges.

pybbn.associational.get_pot(nodes: list[Any], domains: dict[Any, list[Any]]) → DataFrame

Create a potential for the specified nodes. All values are initialized to 1.

Parameters:

nodes – Nodes.
domains – Domains.

Returns:

Potential.

pybbn.associational.get_required_edges(g: Graph, node: Any) → list[tuple[Any, Any]]

Get the required edges that will be created by connecting all the neighbors of node. This implementation avoids constructing an intermediate list of all combinations which becomes costly for high-degree nodes.

Parameters:

g – Graph (moralized graph derived).
node – Node.

Returns:

List of missing edges between neighbors of node.

pybbn.associational.get_triangulation_info(g: Graph, n: Any, domains: dict[Any, list[Any]]) → dict[str, Any]

Get the triangulation information for the given node.

Parameters:

g – Graph (moralized).
n – Node.
domains – Domains.

Returns:

Triangulation information (e.g. weight and edges).

pybbn.associational.init_join_tree(clusters: list[set[Any]]) → Graph

Initialize a join tree. Nodes are the clusters. There are no edges. There are no separation sets.

Parameters:: clusters – Clusters.
Returns:: Join tree (or forest).

pybbn.associational.is_subset(lhs: set[Any], clusters: list[set[Any]]) → bool

Check if the specified cluster (lhs) is subset of the ones in the list of clusters.

Parameters:

lhs – Cluster.
clusters – Clusters.

Returns:

True if the specified cluster (lhs) is subset of the clusters, False otherwise.

pybbn.associational.marginalize(P: DataFrame, nodes: list[str]) → DataFrame

Marginalize the potential keeping only the nodes specified.

Parameters:

P – Potential.
nodes – Nodes to keep.

Returns:

Potential.

pybbn.associational.marginalize_np(P: ArrayPotential, nodes: list[str]) → ArrayPotential: Marginalize an array potential keeping only the provided nodes.

pybbn.associational.moralize(d: DiGraph, u: Graph) → Graph

Moralize the DAG.

Parameters:

d – DAG.
u – Undirected graph.

Returns:

Moralized graph.

pybbn.associational.multiply(L: DataFrame, R: DataFrame, columns: list[str] | None = None) → DataFrame

Multiply two potentials L x R. The resulting product has the dimension of R.

Parameters:

L – Potential.
R – Potential.
columns – Precomputed common nodes between L and R. If None then the common nodes are computed on the fly.

Returns:

Potential.

pybbn.associational.multiply_np(L: ArrayPotential, R: ArrayPotential) → ArrayPotential: Multiply two array potentials using numpy.einsum.

pybbn.associational.normalize(P: DataFrame) → DataFrame

Normalize the potential.

Returns:: Potential.

pybbn.associational.normalize_np(P: ArrayPotential) → ArrayPotential: Normalize an array potential.

pybbn.associational.potential_to_array(p: DataFrame, domains: dict[Any, list[Any]]) → ArrayPotential: Convert a DataFrame potential to an ArrayPotential.

pybbn.associational.send_message(X: DataFrame, s: DataFrame, Y: DataFrame, xs_nodes: list[str] | None = None, sy_nodes: list[str] | None = None) → tuple[DataFrame, DataFrame, DataFrame]

Send message from X -> s -> Y.

Parameters:

X – Potential.
s – Potential.
Y – Potential.
xs_nodes – Precomputed common nodes between X and s.
sy_nodes – Precomputed common nodes between s and Y.

Returns:

New potentials (X, s, Y) as a result of message pass.

pybbn.associational.send_message_np(X: ArrayPotential, s: ArrayPotential, Y: ArrayPotential, xs_axes: list[str] | None = None) → tuple[ArrayPotential, ArrayPotential, ArrayPotential]: Send a message from X -> s -> Y using array potentials.

pybbn.associational.triangulate(m: Graph, domains: dict[Any, list[Any]]) → tuple[Graph, list[set[Any]]]

Triangulate the moralized graph.

Parameters:

m – Moralized graph.
domains – Domains.

Returns:

Tuple of the triangulated graph and induced clusters during triangulation.

Interventional Query

pybbn.interventional.do_by_backdoor_adjustment(Y: list[Any], y: list[str], X: list[Any], x: list[str], Z: list[Any], model: IReasoningModel) → Series

Perform the do-operation using the (backdoor) adjustment formula.

\(P(Y=y|do(X=x)) = \sum_{z \in Z} P(Y=y|X=x, Z=z)P(Z=z)\)

The output is the causal effect. To estimate the average causal effect, call this function twice to compute the following quantity.

\(P(Y=y|do(X=x_1)) - P(Y=y|do(X=x_2))\)

Parameters:

Y – Set of variables Y.
y – Set of values corresponding to Y.
X – Set of variables X.
x – Set of values corresponding to X.
Z – Set of variables Z. Should be confounders.
model – Reasoning model.

Returns:

\(P(Y=y|do(X=x))\)

pybbn.interventional.do_by_graph_surgery(Y: list[Any], y: list[str], X: list[Any], x: list[str], model: IReasoningModel) → Series

Performs the do-operation using doing graph surgery.

Parameters:

Y – Set of variables Y.
y – Set of values corresponding to Y.
X – Set of variables X.
x – Set of values corresponding to X.
model – Reasoning model.

Returns:

\(P(Y=y|do(X=x))\)

pybbn.interventional.get_domain_product(nodes: list[Any], domains: dict[Any, list[Any]]) → list[list[Any]]

Get product of domains.

Parameters:

nodes – Nodes.
domains – Domains.

Returns:

Product of node domains.

Counterfactual Query

pybbn.counterfactual.create_potential(y: Any, p: DataFrame, u_i: int, wpt: ndarray) → DataFrame

Creates a new potential associated with the hidden variable state.

Parameters:

y – Node.
p – Potential.
u_i – Hidden variable state.
wpt – Wiggled probabilities.

Returns:

Potential.

pybbn.counterfactual.create_potential_with_hidden(y: Any, df: DataFrame, u: Any | None = None, max_wiggles: int = 100) → DataFrame

Creates a potential associated with a hidden variable.

Parameters:

y – Node.
df – Potential.
u – Noise generator.
max_wiggles – Maximum number of wiggles. Defaults to 100.

Returns:

Potential with hidden state.

pybbn.counterfactual.get_cpt(y: Any, pot: DataFrame) → ndarray

Get CPT from potential.

Parameters:

y – Node.
pot – Potential.

Returns:

CPT.

pybbn.counterfactual.get_scm_dag(y: Any, model: IReasoningModel, all_hidden: bool = False) → DiGraph

Get SCM DAG. Equivalent to SCM structural function.

Parameters:

y – Node.
model – Sampling model.
all_hidden – Whether to associate all nodes with hidden variables. Defaults to False.

Returns:

SCM DAG with respect to y.

pybbn.counterfactual.get_scm_dp(y: Any, model: IReasoningModel, u: Any | None = None, max_wiggles: int = 100, all_hidden: bool = False) → tuple[DiGraph, dict[str, Any]]

Get SCM DAG and parameters.

Parameters:

y – Node.
model – Sampling model.
u – List of noise generator. Should only be 2 at most. If only 1 is defined, then it is recycled.
max_wiggles – Maximum number of noise samples.
all_hidden – Whether to associate all nodes with hidden variable. Default is False.

Returns:

SCM equation (tuple of graph and parameters).

pybbn.counterfactual.get_scm_parameters(y: Any, model: IReasoningModel, u: Any | None = None, max_wiggles: int = 100, all_hidden: bool = False) → dict[Any, list[dict[str, Any]] | dict[str, Any]]

Get SCM parameters.

Parameters:

y – Node.
model – Sampling model.
u – Noise generator.
max_wiggles – Maximum number of wiggles. Defaults to 100.
all_hidden – Whether to associate all nodes with hidden variables. Defaults to False.

Returns:

CPTs.

pybbn.counterfactual.softmax(p: ndarray) → ndarray

Normalize probabilities using softmax.

Parameters:: p – Probabilities.
Returns:: Normalized probabilities.

pybbn.counterfactual.wiggle_cpt(cpt: ndarray, u: Any, max_wiggles: int = 100) → ndarray

Wiggles the CPT.

Parameters:

cpt – CPT.
u – Noise.
max_wiggles – Maximum number of wiggles. Defaults to 100.

Returns:

Wiggled CPT.

pybbn.counterfactual.wiggle_p(p: ndarray, u: Any, max_wiggles: int = 100) → ndarray

Wiggle probabilities the specified number of times.

Parameters:

p – Probabilities.
u – Noise generator.
max_wiggles – Maximum number of wiggles. Defaults to 100.

Returns:

Wiggled probabilities.

pybbn.counterfactual.wiggle_p_once(p: ndarray, u: Any) → ndarray

Wiggle probabilities.

Parameters:

p – Probabilities.
u – Noise generator.

Returns:

Wiggled probabilities.

Graphical Query

pybbn.graphical.do(d: DiGraph, X: list[Any]) → DiGraph

Perform surgery on a DAG by remove links from parents to children. The children are the set of nodes specified by X.

Parameters:

d – DAG.
X – List of nodes.

Returns:

Intervened graph.

pybbn.graphical.find_minimal_confounders(g: GraphTuple, path: list[Any], x: Any, y: Any, z: list[Any]) → list[Any]

Find a minimal set of confounders in the path.

Parameters:

g – GraphTuple.
path – Path.
x – X node.
y – Y node.
z – Z node.

Returns:

Minimal set of confounders.

pybbn.graphical.find_minimal_mediator(g: GraphTuple, path: list[Any], x: Any, y: Any, z: list[Any]) → list[Any]

Find a minimal set of mediators in the path.

Parameters:

g – GraphTuple.
path – Path.
x – X node.
y – Y node.
z – Z node.

Returns:

Minimal set of mediators in the path.

pybbn.graphical.get_all_confounders(g: GraphTuple, x: Any, y: Any) → set[Any]

Get all confounders.

Parameters:

g – GraphTuple.
x – X node.
y – Y node.

Returns:

Set of confounders.

pybbn.graphical.get_all_mediators(g: GraphTuple, x: Any, y: Any) → set[Any]

Get all mediators.

Parameters:

g – GraphTuple.
x – X node.
y – Y node.

Returns:

Set of mediators.

pybbn.graphical.get_all_paths(g: GraphTuple, start: Any, stop: Any) → list[tuple[Any]]

Get all paths from start to stop.

Parameters:

g – GraphTuple.
start – Start node.
stop – Stop node.

Returns:

All simple paths between start and stop nodes.

pybbn.graphical.get_graph_colliders(g: GraphTuple, path: list[Any]) → list[Any]

Get a list of nodes in the path that are colliders in graph.

Parameters:

g – GraphTuple.
path – Path.

Returns:

List of colliders.

pybbn.graphical.get_graph_tuple(d: DiGraph) → GraphTuple

Get the graph tuple.

Parameters:: d – DAG.
Returns:: GraphTuple.

pybbn.graphical.get_minimal_confounders(g: GraphTuple, x: Any, y: Any) → list[Any]

Find a minimal set of confounders in the graph with respect to x and y.

Parameters:

g – GraphTuple.
x – X node.
y – Y node.

Returns:

Minimal set of confounders in the graph with respect to x and y.

pybbn.graphical.get_minimal_mediators(g: GraphTuple, x: Any, y: Any) → list[Any]

Find a minimal set of mediators in the graph with respect to x and y.

Parameters:

g – GraphTuple.
x – X node.
y – Y node.

Returns:

Minimal set of mediators in the graph with respect to x and y.

pybbn.graphical.get_path_colliders(g: GraphTuple, path: list[Any]) → list[Any]

Get a list of nodes that are colliders in the path.

Parameters:

g – GraphTuple.
path – Path.

Returns:

List of colliders.

pybbn.graphical.get_path_triplets(path: list[Any]) → list[list[Any]]

Get all path contiguous triplets from start to stop.

Parameters:: path – Path.
Returns:: List of contiguous triplets.

pybbn.graphical.get_paths(g: GraphTuple, x: Any, y: Any) → list[dict[str, Any]]

Get a list of paths with metadata about each path.

Parameters:

g – GraphTuple.
x – X node.
y – Y node.

Returns:

List of paths with metadata.

pybbn.graphical.get_triplet_type(g: GraphTuple, x: Any, z: Any, y: Any) → str

Get triplet type. Can be serial, diverging or converging.

Parameters:

g – GraphTuple.
x – X node.
z – Z node.
y – Y node.

Returns:

Triplet type.

pybbn.graphical.get_twin_network(d: DiGraph) → DiGraph

Get a twin network. Let, D, be a dag, and D’ be an identical copy of D, then a twin network is defined as a DAG composed of D and D’ where the same/corresponding nodes, N_i and N’_i between D and D’, are connected by a latent variable U_i.

Parameters:: d – DAG.
Returns:: Twin network.

pybbn.graphical.get_undirected_graph(d: DiGraph) → Graph

Get the undirected graph.

Parameters:: d – DAG.
Returns:: Undirected graph.

pybbn.graphical.is_active_path(g: GraphTuple, path: list[Any], evidence: set[Any] | None = None) → bool

Check if a path is active.

Parameters:

g – GraphTuple.
path – Path.
evidence – Set of evidence.

Returns:

True if the path is active, False otherwise.

pybbn.graphical.is_active_triplet(g: GraphTuple, x: Any, z: Any, y: Any, evidence: set[Any] | None = None) → bool

Check if x, z, y configuration is active. Active and inactive depends on the configuration and whether there is any evidence.

Parameters:

g – GraphTuple.
x – X node.
z – Z node.
y – Y node.
evidence – Set of evidence.

Returns:

True if the triplet is active, False otherwise.

pybbn.graphical.is_any_collider(g: GraphTuple, z: Any) → bool

Check if z is in any triplet configuration where it is a collider.

Parameters:

g – GraphTuple.
z – Z node.

pybbn.graphical.is_d_separated(g: GraphTuple, x: Any, y: Any, evidence: set[Any] | None = None) → bool

Check if x and y are d-separated given the evidence.

Parameters:

g – GraphTuple.
x – X node.
y – Y node.
evidence – Set of evidence.

Returns:

True if x and y are d-separated given the evidence, False otherwise.

Sampling

pybbn.sampling.get_cdf(model: IReasoningModel) → dict[Any, Any]

Get CDF for nodes.

Parameters:: model – Sampling model.
Returns:: CDFs.

pybbn.sampling.sample(model: IReasoningModel, max_samples: int = 100, evidence: dict[Any, Any] | None = None, seed: int = 37) → DataFrame

Applies logic sampling to generate samples.

Parameters:

model – Sampling model.
max_samples – Maximum number of samples.
evidence – Any evidence.
seed – Seed. Default is 37.

Returns:

Samples.

pybbn.sampling.sample_twin(model: IReasoningModel, max_samples: int = 100, p: float = 0.5, seed: int = 37) → DataFrame

Sample data for twin network. Every node should have a corresponding twin node and noise node.

Parameters:

model – Model.
max_samples – Maximum number of samples.
p – Parameter for binomial sample.
seed – Seed. Default is 37.

Returns:

Samples.

Generator

class pybbn.generator.NodeParams

Bases: TypedDict

Node params.

columns: list[str | float]

data: list[list[str | float]]

pybbn.generator.generate_multi_bbn(n: int, max_iter: int = 10, max_values: int = 2, max_alpha: int = 10, rng: Generator | None = None) → tuple[DiGraph, dict[str, Any]]

Generates structure and parameters for a multi-connected BBN.

Parameters:

n – Number of nodes.
max_iter – Maximum iterations.
max_values – Maximum values per node.
max_alpha – Maximum alpha per value (hyperparameters).
rng – Random number generator.

Returns:

A tuple of structure and parameters.

pybbn.generator.generate_singly_bbn(n: int, max_iter: int = 10, max_values: int = 2, max_alpha: int = 10, rng: Generator | None = None) → tuple[DiGraph, dict[str, Any]]

Generates structure and parameters for a singly-connected BBN.

Parameters:

n – Number of nodes.
max_iter – Maximum iterations.
max_values – Maximum values per node.
max_alpha – Maximum alpha per value (hyperparameters).
rng – Random number generator.

Returns:

A tuple of structure and parameters.

Serde

pybbn.serde.dict_to_graph(d: dict[str, Any], graph_type: str) → Graph | DiGraph

Convert a dictionary to a graph.

Parameters:

d – Dictionary.
graph_type – Graph type (“dag” or “undirected”).

Returns:

nx.Graph or nx.DiGraph.

pybbn.serde.dict_to_model(data: dict[str, Any]) → ReasoningModel

Convert dictionary to reasoning model.

Parameters:: data – Dictionary.
Returns:: Reasoning model.

pybbn.serde.dict_to_tree(d: dict[str, Any]) → Graph

Convert dictionary to tree.

Parameters:: d – Dictionary.
Returns:: Tree.

pybbn.serde.from_bbn_file(path: str) → IReasoningModel

Load BBN model from file.

Parameters:: path – Path to file.
Returns:: Reasoning model.

pybbn.serde.graph_to_dict(g: Graph | DiGraph) → dict[str, Any]

Convert graph to dictionary.

Parameters:: g – nx.Graph or nx.DiGraph.
Returns:: Dictionary.

pybbn.serde.model_to_dict(model: ReasoningModel) → dict[str, Any]

Convert a reasoning model to a dictionary.

Parameters:: model – Reasoning model.
Returns:: Dictionary.

pybbn.serde.tree_to_dict(g: Graph) → dict[str, Any]

Convert tree to dictionary.

Parameters:: g – Tree.
Returns:: Dictionary.

External

R

pybbn.ext.r.get_script(model: IReasoningModel | ReasoningModel) → str

Get R script to create model.

Parameters:: model – Reasoning model.
Returns:: R script.

Netica

pybbn.ext.netica.get_script(model: ReasoningModel | IReasoningModel) → str

Get Netica DNE script representing the model.

Parameters

model: ReasoningModel | IReasoningModel: The model to convert to a Netica DNE script.

Returns

str: Script representing the Bayesian network in Netica DNE format.