Functional Bayesian Network¶

class pgmpy.models.FunctionalBayesianNetwork.FunctionalBayesianNetwork(ebunch: List[Tuple[Hashable, Hashable]] | None = None, latents: Set[Hashable] = {}, lavaan_str: str | None = None, dagitty_str: str | None = None)[source]¶

Class for representing Functional Bayesian Network.

Functional Bayesian Networks allow for representation of any probability distribution using CPDs in functional form (Functional CPD). Functional CPDs return a pyro.distribution object allowing for flexible representation of any distribution.

add_cpds(*cpds: FunctionalCPD) → None[source]¶

Adds FunctionalCPDs to the Bayesian Network.

Parameters:: cpds (instances of FunctionalCPD) – List of FunctionalCPDs which will be associated with the model

Examples

>>> from pgmpy.factors.hybrid import FunctionalCPD
>>> from pgmpy.models import FunctionalBayesianNetwork
>>> import pyro.distributions as dist
>>> import numpy as np

>>> model = FunctionalBayesianNetwork([("x1", "x2"), ("x2", "x3")])
>>> cpd1 = FunctionalCPD("x1", lambda _: dist.Normal(0, 1))
>>> cpd2 = FunctionalCPD(
...     "x2", lambda parent: dist.Normal(parent["x1"] + 2.0, 1), parents=["x1"]
... )
>>> cpd3 = FunctionalCPD(
...     "x3", lambda parent: dist.Normal(parent["x2"] + 0.3, 2), parents=["x2"]
... )
>>> model.add_cpds(cpd1, cpd2, cpd3)

check_model() → bool[source]¶

Checks the model for various errors. This method checks for the following error -

Checks if the CPDs associated with nodes are consistent with their parents.

Returns:: check – True if all the checks pass.
Return type:: boolean

fit(data: ~pandas.core.frame.DataFrame, method: str = 'SVI', optimizer: ~pyro.optim.optim.PyroOptim = <pyro.optim.optim.PyroOptim object>, prior_fn: ~typing.Callable | None = None, num_steps: int = 1000, seed: int | None = None, nuts_kwargs: ~typing.Dict | None = None, mcmc_kwargs: ~typing.Dict | None = None) → Dict[str, Any][source]¶

Fit the Bayesian network to data using Pyro’s stochastic variational inference.

Parameters:

data (pandas.DataFrame) – DataFrame with observations of variables.
method (str (default: "SVI")) – Fitting method to use. Currently supports “SVI” and “MCMC”.
optimizer (Instance of pyro optimizer (default: pyro.optim.Adam({"lr": 1e-2}))) – Only used if method is “SVI”. The optimizer to use for optimization.
prior_fn (function) – Only used if method is “MCMC”. A function that returns a dictionary of pyro distributions for each parameter in the model.
num_steps (int (default: 100)) – Number of optimization steps. For SVI it is the num_steps argument for pyro.infer.SVI. For MCMC, it is the num_samples argument for pyro.infer.MCMC.
seed (int (default: None)) – Seed value for random number generator.
nuts_kwargs (dict (default: None)) – Only used if method is “MCMC”. Additional arguments to pass to pyro.infer.NUTS.
mcmc_kwargs (dict (default: None)) – Only used if method is “MCMC”. Additional arguments to pass to pyro.infer.MCMC.

Returns:

dict – If method is “MCMC”, returns a dictionary of posterior samples for each parameter.

Return type:

If method is “SVI”, returns a dictionary of parameter values.

Examples

>>> from pgmpy.factors.hybrid import FunctionalCPD
>>> from pgmpy.models import FunctionalBayesianNetwork
>>> import numpy as np
>>> import pyro.distributions as dist

>>> model = FunctionalBayesianNetwork([("x1", "x2")])
>>> x1 = np.random.normal(0.2, 0.8, size=10000)
>>> x2 = np.random.normal(0.6 + x1, 1)
>>> data = pd.DataFrame({"x1": x1, "x2": x2})

>>> def x1_fn(parents):
...     mu = pyro.param("x1_mu", torch.tensor(1.0))
...     sigma = pyro.param(
...         "x1_sigma", torch.tensor(1.0), constraint=constraints.positive
...     )
...     return dist.Normal(mu, sigma)
...

>>> def x2_fn(parents):
...     intercept = pyro.param("x2_inter", torch.tensor(1.0))
...     sigma = pyro.param(
...         "x2_sigma", torch.tensor(1.0), constraint=constraints.positive
...     )
...     return dist.Normal(intercept + parents["x1"], sigma)
...

>>> cpd1 = FunctionalCPD("x1", fn=x1_prior)
>>> cpd2 = FunctionalCPD("x2", fn=x2_prior, parents=["x1"])
>>> model.add_cpds(cpd1, cpd2)
>>> params = model.fit(data, method="SVI", num_steps=100)
>>> print(params)

>>> def prior_fn():
...     return {
...         "x1_mu": dist.Uniform(0, 1),
...         "x1_sigma": dist.HalfNormal(5),
...         "x2_inter": dist.Normal(1.0),
...         "x2_sigma": dist.HalfNormal(1),
...     }
...

>>> def x1_fn(priors, parents):
...     return dist.Normal(priors["x1_mu"], priors["x1_sigma"])
...

>>> def x2_fn(priors, parents):
...     return dist.Normal(
...         priors["x2_inter"] + parent["x1"], priors["x2_sigma"]
...     )
...

>>> cpd1 = FunctionalCPD("x1", fn=x1_fn)
>>> cpd2 = FunctionalCPD("x2", fn=x2_fn, parents=["x1"])
>>> model.add_cpds(cpd1, cpd2)

>>> params = model.fit(data, method="MCMC", prior_fn=prior_fn, num_steps=100)
>>> print(params["x1_mu"].mean(), params["x1_std"].mean())

get_cpds(node: Any | None = None) → List[FunctionalCPD] | FunctionalCPD[source]¶

Returns the cpd of the node. If node is not specified returns all the CPDs that have been added till now to the graph

Parameter¶

node: any hashable python object (optional): The node whose CPD we want. If node not specified returns all the CPDs added to the model.

rtype:: A list of Functional CPDs.

Examples

>>> from pgmpy.factors.hybrid import FunctionalCPD
>>> from pgmpy.models import FunctionalBayesianNetwork
>>> import numpy as np
>>> import pyro.distributions as dist

>>> model = FunctionalBayesianNetwork([("x1", "x2"), ("x2", "x3")])
>>> cpd1 = FunctionalCPD("x1", lambda _: dist.Normal(0, 1))
>>> cpd2 = FunctionalCPD(
...     "x2", lambda parent: dist.Normal(parent["x1"] + 2.0, 1), parents=["x1"]
... )
>>> cpd3 = FunctionalCPD(
...     "x3", lambda parent: dist.Normal(parent["x2"] + 0.3, 2), parents=["x2"]
... )
>>> model.add_cpds(cpd1, cpd2, cpd3)
>>> model.get_cpds()

remove_cpds(*cpds: FunctionalCPD) → None[source]¶

Removes the given cpds from the model.

Parameters:: *cpds (FunctionalCPD objects) – A list of FunctionalCPD objects that need to be removed from the model.

Examples

>>> from pgmpy.factors.hybrid import FunctionalCPD
>>> from pgmpy.models import FunctionalBayesianNetwork
>>> import numpy as np
>>> import pyro.distributions as dist

>>> model = FunctionalBayesianNetwork([("x1", "x2"), ("x2", "x3")])
>>> cpd1 = FunctionalCPD("x1", lambda _: dist.Normal(0, 1))
>>> cpd2 = FunctionalCPD(
...     "x2", lambda parent: dist.Normal(parent["x1"] + 2.0, 1), parents=["x1"]
... )
>>> cpd3 = FunctionalCPD(
...     "x3", lambda parent: dist.Normal(parent["x2"] + 0.3, 2), parents=["x2"]
... )
>>> model.add_cpds(cpd1, cpd2, cpd3)
>>> for cpd in model.get_cpds():
...     print(cpd)
...

>>> model.remove_cpds(cpd2, cpd3)
>>> for cpd in model.get_cpds():
...     print(cpd)
...

simulate(n_samples: int = 1000, seed: int | None = None) → DataFrame[source]¶

Simulate samples from the model.

Parameters:

n_samples (int, optional (default: 1000)) – Number of samples to generate
seed (int, optional) – The seed value for the random number generator.

Returns:

Simulated samples with columns corresponding to network variables

Return type:

pandas.DataFrame

Examples

>>> from pgmpy.factors.hybrid import FunctionalCPD
>>> from pgmpy.models import FunctionalBayesianNetwork
>>> import numpy as np
>>> import pyro.distributions as dist

>>> model = FunctionalBayesianNetwork([("x1", "x2"), ("x2", "x3")])
>>> cpd1 = FunctionalCPD("x1", lambda _: dist.Normal(0, 1))
>>> cpd2 = FunctionalCPD(
...     "x2", lambda parent: dist.Normal(parent["x1"] + 2.0, 1), parents=["x1"]
... )
>>> cpd3 = FunctionalCPD(
...     "x3", lambda parent: dist.Normal(parent["x2"] + 0.3, 2), parents=["x2"]
... )
>>> model.add_cpds(cpd1, cpd2, cpd3)
>>> model.simulate(n_samples=1000)

Functional Bayesian Network¶

Parameter¶

Navigation

Related Topics