Functional Bayesian Network

class pgmpy.models.FunctionalBayesianNetwork.FunctionalBayesianNetwork(ebunch=None, latents={}, lavaan_str=None, dagitty_str=None)

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)

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

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, method='SVI', optimizer=<pyro.optim.optim.PyroOptim object>, prior_fn=None, num_steps=1000, seed=None, nuts_kwargs=None, mcmc_kwargs=None)

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=None)

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)

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=1000, seed=None)

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)