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)

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

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) → 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]) → 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, 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.

Returns

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

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: [Optional[List[Any]]] = None, evidences: [Optional[Dict[Any, pandas.core.frame.DataFrame]]] = 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=100, evidence=None, seed=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=100, seed=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: Union[DiGraph, Dict[str, Any]], p: Dict[Any, Union[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])

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) → 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) → 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, y)

Get the all confounders wrt to x and y.

Parameters

• x – Node.

• y – Node.

Returns

List of all confounders.

get_all_mediators(x, y)

Get the all mediators wrt to x and y.

Parameters

• x – Node.

• y – Node

Returns

List of all mediators.

get_minimal_confounders(x, y) → 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, y)

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='backdoor') → 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, 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.

• method – Method used to perform the do-operation. Must be one of ‘backdoor’ or ‘graphical’.

Returns

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

is_d_separated(x: Any, y: Any, evidence: Set[Any] = 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: [Optional[List[Any]]] = None, evidences: [Optional[Dict[Any, pandas.core.frame.DataFrame]]] = 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=100, evidence=None, seed=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=100, seed=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) → Any

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

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, Union[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]])

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) → DataFrame

Divide two potentials N / D. The N potential is in the numerator. The D potential is in the denominator.

Parameters

• N – Potential.

• D – Potential.

Returns

Divided potential.

pybbn.associational.dwalk(x: Any, g: Graph, marked: List[Any], messages: List[Tuple[Any, Any, Any]])

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_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 the node.

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 the node.

Parameters

• g – Graph (moralized graph derived).

• node – Node.

Returns

List of edges.

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.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) → DataFrame

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

Parameters

• L – Potential.

• R – Potential.

Returns

Potential.

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

Normalize the potential.

Returns

Potential.

pybbn.associational.send_message(X: DataFrame, s: DataFrame, Y: DataFrame) → Tuple[DataFrame, DataFrame, DataFrame]

Send message from X -> s -> Y.

Parameters

• X – Potential.

• s – Potential.

• Y – Potential.

Returns

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

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=None, max_wiggles=100)

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=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=None, max_wiggles=100, all_hidden=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=None, max_wiggles=100, all_hidden=False) → Dict[Any, Union[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, u, max_wiggles=100)

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, u, max_wiggles=100)

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])

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: 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: 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) → 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=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=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) → 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=100, evidence=None, seed=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=100, p=0.5, seed=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.

Serde

pybbn.serde.dict_to_graph(d: Dict[str, Any], graph_type) → networkx.classes.graph.Graph | networkx.classes.digraph.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: networkx.classes.graph.Graph | networkx.classes.digraph.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.