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

Initialize the graph tuple.

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 ccondquery(target: NodeId, given: list[NodeId] | None = None, *, e: dict[NodeId, Any], h: dict[NodeId, Any], easy: bool = False) → Potential | pd.DataFrame

Query the exact counterfactual conditional distribution \(P(target_h \mid given_h, e)\).

Parameters:

target – Counterfactual target variable.
given – Counterfactual conditioning variables.
e – Factual evidence.
h – Hypothetical intervention.
easy – When True, return a pandas DataFrame.

Returns:

Exact counterfactual conditional distribution.

abstract cjquery(nodes: list[NodeId], *, e: dict[NodeId, Any], h: dict[NodeId, Any], easy: bool = False) → Potential | pd.DataFrame

Query the exact counterfactual joint distribution over nodes.

Parameters:

nodes – Nodes to include in the counterfactual joint distribution.
e – Factual evidence.
h – Hypothetical intervention.
easy – When True, return a pandas DataFrame.

Returns:

Exact counterfactual joint distribution.

abstract property cluster_potentials: Mapping[Any, ArrayPotential]

Cluster potentials.

Returns:: Cluster potentials.

abstract condquery(target: NodeId, given: list[NodeId] | None = None, evidences: PotentialMap | None = None, *, easy: bool = False) → Potential | pd.DataFrame

Query the exact conditional distribution \(P(target \mid given, evidence)\).

Parameters:

target – Target variable.
given – Conditioning variables.
evidences – Evidence potentials.
easy – When True, return a pandas DataFrame instead of an array potential.

Returns:

Conditional distribution with axes [target, *given].

abstract cpevidence(assignments: dict[Any, Any] | None = None, *, e: dict[Any, Any], h: dict[Any, Any]) → float

Compute the exact counterfactual probability of assignments conditioned on factual e.

Parameters:

assignments – Counterfactual-view assignments of interest.
e – Factual evidence.
h – Hypothetical intervention.

Returns:

Exact conditional counterfactual probability.

abstract cpquery(nodes: list[NodeId] | None = None, *, e: dict[NodeId, Any], h: dict[NodeId, Any], easy: bool = False) → PotentialMap | Mapping[NodeId, pd.DataFrame]

Query exact counterfactual marginals under hypothetical h conditioned on factual e.

Parameters:

nodes – Nodes to query in the counterfactual view. Defaults to all model nodes.
e – Factual evidence.
h – Hypothetical intervention.
easy – When True, return pandas DataFrames.

Returns:

Counterfactual marginals for the requested nodes.

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

Counterfactual query.

Compute the counterfactual distribution \(P(Y_h \mid e)\).

Implementations may compile an exact twin network or an equivalent exact structural model, but the returned result should represent the posterior distribution over Y under the hypothetical h after conditioning on the factual evidence e.

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) → ArrayPotential

Create finding evidence. The values should be either 1 or 0. When values is not specified, the evidence order should match the sorted domain values of the node.

Parameters:

n – Node.
evidences – Evidence weights.
values – The domain values corresponding to evidences.

Returns:

Potential.

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

Create observation evidence. 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: Mapping[Any, list[Any]]

Return the variable domains.

Returns:: Domains.

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

Shortcut to create observation evidences.

Parameters:: evidences – Observation evidence assignments.
Returns:: Evidence mapping.

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

Create observation evidence mappings from key-value pairs.

Parameters:: evidences – Observation evidence assignments.
Returns:: Evidence mapping.

abstract property graph_tuple: GraphTuple

Graph tuple.

Returns:: Graph tuple.

abstract intervene(interventions: dict[Any, Any]) → IReasoningModel

Compile the exact intervened model induced by do(interventions).

Parameters:: interventions – Mapping of intervened nodes to their clamped values.
Returns:: Exact intervened reasoning model.

abstract iquery(Y: list[Any], y: list[str], X: list[Any], x: list[str], method: str = 'auto', x_ref: list[str] | None = None, adjustment_set: list[Any] | None = None) → dict[str, float]

Interventional query.

Compute \(P(Y=y\mid do(X=x))\) with exact causal inference.

Implementations may use exact backdoor adjustment, \(P(Y=y|do(X=x)) = \sum_{z \in Z} P(Y=y|X=x, Z=z)P(Z=z)\), or exact graph surgery on an intervened model.

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.
method – Query method. Implementations should prefer an exact path by default and may expose validated shortcuts.
x_ref – Optional reference intervention values for X. When provided, the returned mapping 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.
adjustment_set – Optional adjustment variables used only by backdoor-style implementations.

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 jquery(nodes: list[NodeId], evidences: PotentialMap | None = None, *, easy: bool = False) → Potential | pd.DataFrame

Query the exact joint posterior over nodes.

Parameters:

nodes – Ordered nodes to include in the joint distribution.
evidences – Evidence potentials.
easy – When True, return a pandas DataFrame instead of an array potential.

Returns:

Joint posterior over nodes.

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: Mapping[Any, ArrayPotential]

Node potentials.

Returns:: Node potentials.

property node_potentials_array: Mapping[Any, ArrayPotential]: Compatibility alias for older call sites.

abstract pevidence(evidences: PotentialMap | None = None) → float

Compute the exact probability of the supplied evidence.

Parameters:: evidences – Evidence potentials.
Returns:: Probability of evidence.

property potentials: dict[str, ndarray]: Return raw node potentials.

abstract pquery(nodes: list[NodeId] | None = None, evidences: PotentialMap | None = None) → PotentialMap

Query the model.

Parameters:

nodes – Nodes to return. If omitted, all nodes are returned.
evidences – Evidence potentials.

Returns:

Potentials.

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

Generate samples from the model.

Implementations may use compiled ancestral sampling for the no-evidence path and likelihood weighting for evidence-conditioned sampling. max_rejections may be retained only for compatibility with older rejection-based behavior.

Parameters:

max_samples – Maximum number of samples.
evidence – Observed evidence.
seed – Seed. Default is 37.
max_rejections – Optional legacy rejection bound retained for compatibility.

Returns:

Samples as a DataFrame.

Raises:

ValueError – If evidence references an unknown node or invalid domain value.
RuntimeError – If sampling cannot produce a valid weighted or accepted sample set for the requested evidence.

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

Sample data from the twin network.

Parameters:

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

Returns:

Samples as a DataFrame.

abstract property t: Graph

Triangulated graph.

Returns:: Triangulated graph.

property topological_order: list[str]: Return topological node order.

abstract property u: Graph

Undirected graph of the DAG.

Returns:: Undirected graph.

pybbn.common.NodeId: alias of Any

pybbn.common.Potential: alias of ArrayPotential

Potential

class pybbn.potential.ArrayPotential(values: ndarray, axes: List[str], axis_domains: dict[str, Sequence[Any]] | None = None)

Bases: object

Lightweight numpy representation of a potential with optional axis metadata.

__init__(values: ndarray, axes: List[str], axis_domains: dict[str, Sequence[Any]] | None = None) → None

axes: List[str]

axis_domains: dict[str, Sequence[Any]] | None = None

property axis_to_pos: dict[str, int]: Map axis name to its integer position in values.

cdf_of(axis: str, given: dict[str, Any] | None = None): Return the cumulative distribution over axis.

clone_with(values: ndarray) → ArrayPotential: Create a new potential sharing axes but different values.

copy() → ArrayPotential: Return a deep copy of the potential.

distribution_of(axis: str, given: dict[str, Any] | None = None): Return the marginal/conditional distribution over axis as a 1D array.

property domain_to_index: dict[str, dict[Any, int]]: Map each domain value to its integer offset along the corresponding axis.

classmethod from_records(axes: Iterable[str], records: Iterable[dict[str, Any]], domains: dict[str, Sequence[Any]]) → ArrayPotential: Build a potential from record dictionaries and axis domains.

property layout_signature: tuple[Any, ...]: Return a hashable signature describing the potential layout.

prob_of(assignments: dict[str, Any]) → float: Return the probability at the fully-specified assignment.

query(*args, **kwargs): Delegate query to the pandas representation.

rename_axes(mapping: dict[str, str]) → ArrayPotential: Return a shallowly-renamed copy of the potential axes and domains.

to_cpt_dict() → dict[str, Any]: Convert the potential to the repository CPT dictionary format.

to_dataframe(): Convert the potential to a pandas DataFrame (cached after first construction).

to_dict(*args, **kwargs): Delegate to_dict to the pandas representation.

to_markdown(*args, **kwargs): Delegate to_markdown to the pandas representation.

to_records() → list[dict[str, Any]]: Convert the potential into a list of dictionaries.

values: ndarray

Factory

pybbn.factory.create_reasoning_model(d: nx.DiGraph | dict[str, Any], p: CptMap, *, zero_policy: str = 'smooth', eps: float = 1e-12) → IReasoningModel

Create reasoning model.

Parameters:

d – DAG. Could be a nx.DiGraph or a dictionary.
p – Parameters (CPTs).
zero_policy – Policy for repairing zero / invalid CPT rows during model construction. One of "strict", "uniform", or "smooth".
eps – Small positive value used when zero_policy="smooth".

Returns:

Reasoning model.

Parameter

pybbn.parameter.get_conditional_cpt(pa: list[Any], ch: Any, data: list[dict[str, Any]]) → dict[str, Any]: Learn conditional CPT from dictionary-based data.

pybbn.parameter.get_marginal_cpt(f: Any, data: list[dict[str, Any]]) → dict[str, Any]: Learn marginal CPT from tabular data represented as dictionaries.

pybbn.parameter.learn_parameters(d: DiGraph, data: Any) → dict[str, dict[str, Any]]: Learn the parameters of each node in the DAG from dictionary rows.

Reasoning

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

Bases: IReasoningModel

Reasoning model.

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

Initialize the reasoning model.

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.

ccondquery(target: NodeId, given: list[NodeId] | None = None, *, e: dict[NodeId, Any], h: dict[NodeId, Any], easy: Literal[False] = False) → Potential
ccondquery(target: NodeId, given: list[NodeId] | None = None, *, e: dict[NodeId, Any], h: dict[NodeId, Any], easy: Literal[True]) → DataFrame: Exact counterfactual conditional distribution under factual e and hypothetical h.

cjquery(nodes: list[NodeId], *, e: dict[NodeId, Any], h: dict[NodeId, Any], easy: Literal[False] = False) → Potential
cjquery(nodes: list[NodeId], *, e: dict[NodeId, Any], h: dict[NodeId, Any], easy: Literal[True]) → DataFrame: Exact counterfactual joint distribution over nodes.

property cluster_potentials: Mapping[Any, ArrayPotential]

Cluster potentials.

Returns:: Cluster potentials.

condquery(target: NodeId, given: list[NodeId] | None = None, evidences: PotentialMap | None = None, *, easy: Literal[False] = False) → Potential
condquery(target: NodeId, given: list[NodeId] | None = None, evidences: PotentialMap | None = None, *, easy: Literal[True]) → DataFrame: Exact conditional distribution P(target | given, evidences).

cpevidence(assignments: dict[Any, Any] | None = None, *, e: dict[Any, Any], h: dict[Any, Any]) → float: Exact counterfactual probability of assignments conditioned on factual e.

cpquery(nodes: list[NodeId] | None = None, *, e: dict[NodeId, Any], h: dict[NodeId, Any], easy: Literal[False] = False) → PotentialMap
cpquery(nodes: list[NodeId] | None = None, *, e: dict[NodeId, Any], h: dict[NodeId, Any], easy: Literal[True]) → Mapping[NodeId, pd.DataFrame]: Exact counterfactual marginals under hypothetical h conditioned on factual e.

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

Counterfactual query.

Compute the exact counterfactual distribution \(P(Y_h \mid e)\).

The implementation compiles an exact twin network in which the factual and counterfactual worlds share exogenous variables derived from the original CPTs. Compiled twin models are cached by intervention signature. The public return value stays array-backed so repeated calls avoid eager pandas materialization; callers that want a dataframe view can still use to_dataframe().

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) → ArrayPotential

Create finding evidence. The values should be either 1 or 0. When values is not specified, the evidence order should correspond to the sorted domain values of the node or variable.

Parameters:

n – Node.
evidences – Evidence weights.
values – The domain values corresponding to evidences.

Returns:

Potential.

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

Create observation evidence. 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) → ArrayPotential

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

Parameters:

n – Node.
evidences – Evidence weights.
values – The domain values corresponding to evidences.

Returns:

Potential.

property d: DiGraph

DAG.

Returns:: DAG.

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

Return the variable domains.

Returns:: Domains.

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

Shortcut to create observation evidences.

Parameters:: evidences – Observation evidence assignments.
Returns:: Evidence mapping.

static from_binary_state(nodes: list[str], domains: dict[str, list[str]], potentials: dict[str, ndarray], parents_by_node: dict[str, list[str]] | None = None) → ReasoningModel: Rebuild reasoning model from binary state components.

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

Get all confounders with respect to x and y.

Parameters:

x – Node.
y – Node.

Returns:

List of all confounders.

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

Get all mediators with respect 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 with respect 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 with respect to x and y.

Parameters:

x – Node.
y – Node.

Returns:

List of minimal mediators.

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

Create observation evidence mappings from key-value pairs.

Parameters:: evidences – Observation evidence assignments.
Returns:: Evidence mapping.

property graph_tuple: GraphTuple

Graph tuple.

Returns:: Graph tuple.

intervene(interventions: dict[Any, Any]) → IReasoningModel: Compile the exact intervened model induced by do(interventions).

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

Interventional query.

Compute \(P(Y=y\mid do(X=x))\) using an exact causal query path.

method="auto" favors correctness first: it uses a cheap observation shortcut only when every intervened node is a root of the DAG, and otherwise falls back to exact graph surgery. method="graph" and method="graphical" always compile an exact intervened model. method="backdoor" remains available for explicit adjustment-based queries, but callers should provide an adjustment set unless the query is the simple single-treatment single-outcome case where minimal confounders can be derived from the graph.

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 "auto", "backdoor", "graph", or "graphical".
x_ref – Optional reference intervention values for X. When provided, the returned mapping equals \(P(Y=y|do(X=x)) - P(Y=y|do(X=x'))\), where \(x'\) corresponds to x_ref.
adjustment_set – Optional adjustment variables used only when method="backdoor".

Returns:

Mapping of variables in Y to \(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 x and y are d-separated; otherwise False.

property j: Graph

Join tree graph.

Returns:: Join tree graph.

jquery(nodes: list[NodeId], evidences: PotentialMap | None = None, *, easy: Literal[False] = False) → Potential

jquery(nodes: list[NodeId], evidences: PotentialMap | None = None, *, easy: Literal[True]) → DataFrame

Exact joint posterior over nodes.

The implementation uses the chain rule over repeated exact single-node posteriors, so it remains exact while supporting arbitrary query node sets with the existing public inference path.

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: Mapping[Any, ArrayPotential]

Node potentials.

Returns:: Node potentials.

pevidence(evidences: PotentialMap | None = None) → float: Exact probability of evidence.

property potentials: dict[str, ndarray]: Return raw node potentials.

pquery(nodes: list[NodeId] | None = None, evidences: PotentialMap | None = None, *, easy: Literal[False] = False) → PotentialMap

pquery(nodes: list[NodeId] | None = None, evidences: PotentialMap | None = None, *, easy: Literal[True]) → Mapping[NodeId, pd.DataFrame]

Probabilistic query.

Parameters:

nodes – Nodes to return. If omitted, all nodes are returned.
evidences – Evidence potentials.
easy – Convenience flag. When False (default) return lightweight ArrayPotential instances; when True materialize pandas DataFrames for ergonomics.

Returns:

Potentials or DataFrames, depending on easy.

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

Generate samples from the model.

The no-evidence path uses compiled ancestral sampling. When evidence is provided, sampling supports both likelihood weighting and rejection sampling.

Parameters:

max_samples – Maximum number of samples.
evidence – Observed evidence.
seed – Seed. Default is 37.
max_rejections – Optional rejection budget used only for method="rejection".
method – Evidence-handling strategy. One of "likelihood" or "rejection".

Returns:

Samples as a DataFrame.

Raises:

ValueError – If evidence references an unknown node or invalid domain value.
RuntimeError – If sampling cannot produce a valid weighted or accepted sample set for the requested evidence.

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

Sample data from the twin network.

Parameters:

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

Returns:

Samples with twin and noise nodes.

scm_equation(node: Any) → IReasoningModel

Return the SCM corresponding to node.

Parameters:: node – Target node.
Returns:: Reasoning model for the SCM.

property t: Graph

Triangulated graph.

Returns:: Triangulated graph.

property topological_order: list[str]: Return topological node order.

property u: Graph

Undirected graph of the DAG.

Returns:: Undirected graph.

Associational Query

pybbn.associational.clone_potential_map(m: Mapping[Any, ArrayPotential]) → Mapping[Any, ArrayPotential]

Clones the potentials.

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

pybbn.associational.cpt_to_pot(node: Any, cpts: dict[Any, ArrayPotential | list[dict[str, Any]] | dict[str, Any]], domains: Mapping[Any, list[Any]], *, domain_to_index: dict[Any, dict[Any, int]] | None = None, shared_axis_domains: dict[Any, tuple[Any, ...]] | None = None) → ArrayPotential

Create potential of the specified node.

Parameters:

node – Node.
cpts – CPTs.
domains – Domain map.

Returns:

Potential.

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

Create finding evidence. Values must be either 0 or 1.

Parameters:

n – Node.
values – Domain (list of values).
evidences – Evidence weights.

Returns:

Potential.

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

Create observation evidence. Exactly one value receives a weight of 1 and the rest receive 0.

Parameters:

n – Node.
values – Domain (list of values).
v – Observed value.

Returns:

Potential.

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

Create all candidate separation-set records for an initialized join tree.

This helper is retained for compatibility and diagnostics. It enumerates every cluster pair with a non-empty intersection, computes the candidate sepset metadata, and returns the records sorted by decreasing sepset mass and increasing cluster cost. Sepset identifiers are canonicalized so the same pair always maps to the same id regardless of traversal order.

Deprecated:: Prefer get_join_tree(), which now builds the join tree directly without materializing and sorting the full candidate list.

Parameters:

g – Initialized join tree.
domains – Domains.

Returns:

List of candidate sepset records sorted by mass descending and cost ascending.

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

Create virtual evidence. Values must be in the range [0, 1].

Parameters:

n – Node.
values – Domain (list of values).
evidences – Evidence weights.

Returns:

Potential.

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

Walk the graph to get the collect message passes using an explicit stack to avoid deep recursion.

Parameters:

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

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: ArrayPotential, D: ArrayPotential) → ArrayPotential

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

Parameters:

N – Potential.
D – Potential.

Returns:

Divided potential with axes matching D.

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

Walk the graph to get the distribute message passes using an explicit stack to avoid deep recursion.

Parameters:

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

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

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

Parameters:

g – Graph (moralized).
domains – Domains.

Returns:

Triangulation information sorted by edges (asc) then weight (asc).

pybbn.associational.get_best_triangulation(g: Graph, domains: Mapping[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: Mapping[Any, list[Any]]) → int

Get the cluster weight.

Parameters:

nodes – Nodes.
domains – Domains.

Returns:

Weight.

pybbn.associational.get_collect_messages(x: Any, g: Graph, transitions: dict[Any, list[tuple[Any, Any]]] | None = None) → list[tuple[Any, Any, Any]]

Get the collect evidence messages.

Parameters:

x – X node.
g – Join tree.
transitions – Optional precomputed cluster transition lookup.

Returns:

List of message pass.

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

Get the common nodes between two potentials.

Parameters:

X – Potential.
Y – Potential.

Returns:

List of common nodes preserving the order of X.

pybbn.associational.get_distribute_messages(x: Any, g: Graph, transitions: dict[Any, list[tuple[Any, Any]]] | None = None) → list[tuple[Any, Any, Any]]

Get the distribute evidence messages.

Parameters:

x – X node.
g – Join tree.
transitions – Optional precomputed cluster transition lookup.

Returns:

List of message pass.

pybbn.associational.get_domains(cpts: dict[Any, ArrayPotential | list[dict[str, Any]] | dict[str, Any]]) → Mapping[Any, list[Any]]

Derive the domains of each variable/node from the CPT specifications.

Parameters:: cpts – Mapping of node identifiers to CPT definitions.
Returns:: Domain map keyed by variable name.

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

Build a join tree from triangulated clusters.

The tree is constructed incrementally from cluster overlap information using a heap ordered by larger shared-member sets first and lower combined cluster cost second. If the cluster-overlap graph is disconnected, the remaining components are bridged with empty sepsets so the returned graph is still a single join tree.

Parameters:

clusters – Clusters.
domains – Domains.

Returns:

Join tree containing cluster and sepset nodes.

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: Mapping[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: Mapping[Any, list[Any]]) → ArrayPotential

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: Mapping[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: ArrayPotential, nodes: list[str]) → ArrayPotential

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: ArrayPotential, R: ArrayPotential) → ArrayPotential

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

Parameters:

L – Potential.
R – Potential.

Returns:

Potential product with axes matching R.

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

Normalize the potential.

Returns:: Potential.

pybbn.associational.normalize_cpt_potential(node: Any, pot: ArrayPotential, *, zero_policy: str = 'smooth', eps: float = 1e-12) → ArrayPotential

Normalize a node CPT and repair invalid rows according to zero_policy.

Policies:

"strict": leave rows untouched.
"uniform": keep positive rows normalized, replace zero / invalid rows with a uniform distribution.
"smooth": replace non-positive entries in positive rows with eps, renormalize, and replace zero / invalid rows with a uniform distribution.

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

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.triangulate(m: Graph, domains: Mapping[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

Perform the do-operation by compiling the exact intervened graph.

The intervened DAG is produced with graph surgery and each affected node is clamped with a deterministic CPT. All untouched nodes reuse the compiled node potentials from the original model, so the query stays on the exact inference path without sampling or parameter relearning.

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: Mapping[Any, list[Any]]) → list[list[Any]]

Get product of domains.

Parameters:

nodes – Nodes.
domains – Domains.

Returns:

Product of node domains.

Counterfactual Query

exception pybbn.counterfactual.CounterfactualSupportError(report: CounterfactualSupportReport)

Bases: ValueError

Raised when exact twin compilation is impossible for the current CPT set.

__init__(report: CounterfactualSupportReport) → None: Store the support report and summarize the first invalid CPT row.

class pybbn.counterfactual.CounterfactualSupportIssue(node: Any, given: dict[Any, Any], reason: str, total: float, row: tuple[float, ...])

Bases: object

Describe one invalid conditional row that blocks exact twin compilation.

__init__(node: Any, given: dict[Any, Any], reason: str, total: float, row: tuple[float, ...]) → None

given: dict[NodeId, Any]

node: NodeId

reason: str

row: tuple[float, ...]

total: float

class pybbn.counterfactual.CounterfactualSupportReport(issue_count: int, issues: tuple[CounterfactualSupportIssue, ...])

Bases: object

Summarize whether exact counterfactual compilation is supported.

__init__(issue_count: int, issues: tuple[CounterfactualSupportIssue, ...]) → None

property is_supported: bool: Return True when exact twin compilation is safe.

issue_count: int

issues: tuple[CounterfactualSupportIssue, ...]

class pybbn.counterfactual.GaussianNoise(mean: float = 0.0, std: float = 1.0, seed: int | None = None)

Bases: object

Simple normal noise source used when SciPy is unavailable.

__init__(mean: float = 0.0, std: float = 1.0, seed: int | None = None) → None: Create a Gaussian noise generator.

rvs(size: tuple[int, ...] | int) → ndarray: Draw random samples from the Gaussian distribution.

pybbn.counterfactual.analyze_counterfactual_support(model: IReasoningModel, max_issues: int = 8) → CounterfactualSupportReport

Inspect the compiled node CPTs for rows that make exact twin compilation invalid.

Exact twin construction requires every conditional row to define a proper distribution. Rows with non-finite entries or non-positive total mass are rejected before the twin network is compiled.

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

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: ArrayPotential, u: Any | None = None, max_wiggles: int = 100) → ArrayPotential

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.ensure_counterfactual_support(model: IReasoningModel, support_report: CounterfactualSupportReport | None = None, max_issues: int = 8) → CounterfactualSupportReport: Validate exact twin support and raise a descriptive error when it is invalid.

pybbn.counterfactual.get_counterfactual_nodes(model: IReasoningModel, target: NodeId | list[NodeId] | tuple[NodeId, ...] | set[NodeId], interventions: dict[NodeId, Any] | None = None) → tuple[NodeId, ...]: Return the nodes that require a counterfactual copy for target.

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

Get CPT from potential.

Parameters:

y – Node.
pot – Potential.

Returns:

CPT.

pybbn.counterfactual.get_exact_twin_dp(model: IReasoningModel, interventions: dict[NodeId, Any] | None = None, *, target: NodeId | list[NodeId] | tuple[NodeId, ...] | set[NodeId] | None = None, support_report: CounterfactualSupportReport | None = None) → tuple[nx.DiGraph, CptMap]: Build an exact twin network and CPTs from the compiled node potentials.

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, ArrayPotential | 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 contiguous triplets along the path.

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[str, Series]: Compute cumulative distribution functions for each node.

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

Sample a model and return assignments as a pandas DataFrame.

The no-evidence path uses a compiled ancestral sampler built from the array-backed node CPTs. When evidence is supplied, callers can choose either likelihood weighting or explicit rejection sampling.

Parameters:

model – Reasoning model to sample from.
max_samples – Number of rows to return.
evidence – Optional observed node assignments.
seed – Seed for the random number generator.
max_rejections – Optional rejection budget used only when method="rejection".
method – Evidence-handling strategy. One of "likelihood" or "rejection".

Returns:

Sampled assignments as a pandas DataFrame.

Raises:

ValueError – If evidence references an unknown node or a value not in that node’s domain.
RuntimeError – If the evidence produces zero total weight or the rejection budget is exhausted.

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

Sample data for the twin network including counterfactual and noise nodes.

Parameters:

model – Reasoning model to sample from.
max_samples – Number of rows to generate.
p – Bernoulli probability used to simulate the exogenous noise nodes.
seed – Seed for the random number generator.

Returns:

Sampled twin-network rows as a pandas DataFrame.

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: np.random.Generator | None = None, max_degree: int | None = None) → tuple[nx.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.
max_degree – Optional maximum number of incident arcs per node.

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: np.random.Generator | None = None, max_degree: int | None = None) → tuple[nx.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.
max_degree – Optional maximum number of incident arcs per node.

Returns:

A tuple of structure and parameters.

Serde

exception pybbn.serde.ModelSchemaError

Bases: ValueError

Raised when a serialized reasoning model uses an unsupported schema version.

pybbn.serde.deserialize_binary(data: bytes) → ReasoningModel: Deserialize reasoning model from simple binary format.

pybbn.serde.dict_to_graph(d: dict[str, Any], graph_type: str) → nx.Graph | nx.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], *, zero_policy: str = 'smooth', eps: float = 1e-12) → ReasoningModel

Convert a serialized dictionary payload to a reasoning model.

The loader accepts legacy payloads without "__version__" as schema version 1. Payloads with a newer schema version than this release supports raise ModelSchemaError.

Parameters:

data – Dictionary.
zero_policy – Policy for repairing zero / invalid CPT rows during model construction. One of "strict", "uniform", or "smooth".
eps – Small positive value used when zero_policy="smooth".

Returns:

Reasoning model.

Raises:

ModelSchemaError – If the payload schema version is invalid or newer than the current implementation supports.

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, *, zero_policy: str = 'smooth', eps: float = 1e-12) → IReasoningModel

Load BBN model from file.

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

pybbn.serde.graph_to_dict(g: nx.Graph | nx.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 versioned dictionary payload.

The returned mapping is suitable for JSON serialization and includes a "__version__" field so future schema changes can be detected during deserialization.

Parameters:: model – Reasoning model.
Returns:: Versioned dictionary representation of the model.

pybbn.serde.serialize_binary(model: IReasoningModel) → bytes: Serialize reasoning model to simple binary format.

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

Return an R script that recreates the model.

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

Netica

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

Return a Netica DNE script for the model.

Parameters

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

Returns

str: A script representing the Bayesian network in Netica DNE format.