tessl/pypi-pgmpy
A library for Probabilistic Graphical Models
—
Pending
models.mddocs/
Models and Graph Structures
Core model classes for creating and managing probabilistic graphical models. pgmpy provides various model types to represent different kinds of probabilistic relationships and graph structures.
Capabilities
Discrete Bayesian Networks
The primary class for working with discrete Bayesian networks - directed acyclic graphs where each node represents a random variable and edges encode conditional dependencies.
class DiscreteBayesianNetwork:
def __init__(self, ebunch=None, latents=set(), lavaan_str=None, dagitty_str=None):
"""
Create a discrete Bayesian network.
Parameters:
- ebunch: list of edges as tuples
- latents: set of latent variable names
- lavaan_str: lavaan model string
- dagitty_str: dagitty model string
"""
def add_edge(self, u, v, **kwargs):
"""Add directed edge from u to v."""
def remove_node(self, node):
"""Remove node and all associated CPDs."""
def add_cpds(self, *cpds):
"""Add conditional probability distributions to the model."""
def get_cpds(self, node=None):
"""Get CPDs for specified node or all CPDs if node is None."""
def remove_cpds(self, *cpds):
"""Remove specified CPDs from the model."""
def get_cardinality(self, node=None):
"""Get cardinality (number of states) for node or all nodes."""
def check_model(self):
"""
Validate model consistency.
Returns:
bool: True if model is valid
Raises:
ValueError: If model has issues
"""
def fit_update(self, data, n_prev_samples=None, n_jobs=1):
"""Update model parameters with new data."""
def predict(self, data, variables=None, n_jobs=1, show_progress=True):
"""
Predict missing values in data.
Parameters:
- data: pandas.DataFrame with missing values
- variables: list of variables to predict
- n_jobs: number of parallel jobs
- show_progress: whether to show progress bar
Returns:
pandas.DataFrame: Data with predicted values
"""
def predict_probability(self, data):
"""Predict probabilities for missing values."""
def get_state_probability(self, states):
"""Get probability of specified variable states."""
def to_markov_model(self):
"""Convert to equivalent Markov network."""
def to_junction_tree(self):
"""Create junction tree for exact inference."""
def simulate(self, n_samples, do=None, evidence=None, show_progress=True):
"""
Generate samples from the model.
Parameters:
- n_samples: number of samples to generate
- do: dict of interventions {variable: value}
- evidence: dict of evidence {variable: value}
- show_progress: whether to show progress bar
Returns:
pandas.DataFrame: Generated samples
"""
def save(self, filename, filetype="bif"):
"""Save model to file in specified format."""
@staticmethod
def load(filename, filetype="bif", **kwargs):
"""Load model from file."""
def copy(self):
"""Create deep copy of the model."""
def get_markov_blanket(self, node):
"""Get Markov blanket of specified node."""
def do(self, nodes, inplace=False):
"""
Perform do-calculus intervention.
Parameters:
- nodes: dict of interventions {variable: value}
- inplace: whether to modify model in-place
Returns:
DiscreteBayesianNetwork: Model with interventions applied
"""Markov Networks
Undirected graphical models representing joint probability distributions through factor graphs.
class MarkovNetwork:
def __init__(self, ebunch=None, latents=[]):
"""
Create a Markov network (undirected graphical model).
Parameters:
- ebunch: list of edges as tuples
- latents: list of latent variable names
"""
def add_edge(self, u, v, **kwargs):
"""Add undirected edge between u and v."""
def add_factors(self, *factors):
"""Add factors (clique potentials) to the model."""
def get_factors(self, factor=None):
"""Get specified factor or all factors."""
def remove_factors(self, *factors):
"""Remove factors from the model."""
def check_model(self):
"""Validate model consistency."""
def copy(self):
"""Create deep copy of the model."""
def to_factor_graph(self):
"""Convert to equivalent factor graph."""
def to_junction_tree(self):
"""Create junction tree for exact inference."""Factor Graphs
Bipartite graphs explicitly representing factorization of probability distributions.
class FactorGraph:
def __init__(self, ebunch=None):
"""
Create a factor graph.
Parameters:
- ebunch: list of edges between variables and factors
"""
def add_edges_from(self, ebunch):
"""Add multiple edges from edge list."""
def add_factors(self, *factors):
"""Add factor nodes to the graph."""
def get_factors(self, factor=None):
"""Get specified factor or all factors."""
def check_model(self):
"""Validate factor graph consistency."""
def to_markov_model(self):
"""Convert to equivalent Markov network."""Junction Trees
Tree structures used for exact inference in graphical models.
class JunctionTree:
def __init__(self, ebunch=None):
"""
Create a junction tree.
Parameters:
- ebunch: list of tree edges
"""
def add_edge(self, u, v, **kwargs):
"""Add edge between cliques."""
def add_factors(self, *factors):
"""Add factors to cliques."""
def check_model(self):
"""Validate junction tree properties."""
def copy(self):
"""Create deep copy of the junction tree."""Dynamic Bayesian Networks
Time-series models representing temporal probabilistic relationships.
class DynamicBayesianNetwork:
def __init__(self, ebunch=None, latents=[]):
"""
Create a Dynamic Bayesian Network for time-series modeling.
Parameters:
- ebunch: list of edges including temporal connections
- latents: list of latent variable names
"""
def add_edge(self, u, v, **kwargs):
"""Add edge, supporting temporal connections."""
def add_cpds(self, *cpds):
"""Add CPDs including temporal dependencies."""
def initialize_initial_state(self, state_names, value):
"""Set initial state distribution."""
def check_model(self):
"""Validate DBN structure and parameters."""
def simulate(self, n_samples, n_time_slices=1):
"""Generate temporal samples from the DBN."""Specialized Models
Additional model types for specific use cases.
class NaiveBayes:
def __init__(self):
"""Create a Naive Bayes classifier model."""
def fit(self, X, y):
"""Fit the model to training data."""
def predict(self, X):
"""Predict class labels for test data."""
def predict_proba(self, X):
"""Predict class probabilities for test data."""
class LinearGaussianBayesianNetwork:
def __init__(self, ebunch=None):
"""Create a Linear Gaussian Bayesian Network."""
def add_cpds(self, *cpds):
"""Add Linear Gaussian CPDs."""
def simulate(self, n_samples):
"""Generate samples from the continuous model."""
class MarkovChain:
def __init__(self, variables=None, card=None, start_state=None):
"""
Create a Markov Chain model.
Parameters:
- variables: list of variable names
- card: list of cardinalities
- start_state: initial state distribution
"""
def add_variable(self, variable, card):
"""Add variable to the chain."""
def add_transition_model(self, transition_model):
"""Add transition probability matrix."""
def sample(self, start_state=None, size=1):
"""Generate samples from the Markov chain."""
# Structural Equation Models
class SEMGraph:
def __init__(self, ebunch=None, latents=[]):
"""Create a Structural Equation Model graph."""
class SEMAlg:
def __init__(self, model):
"""SEM algorithm implementation."""
class SEM:
def __init__(self, ebunch=None, latents=[]):
"""Complete Structural Equation Model."""Graph Base Classes
Fundamental graph structures used by all model types.
class DAG:
def __init__(self, ebunch=None):
"""
Directed Acyclic Graph base class.
Parameters:
- ebunch: list of directed edges
"""
def add_edge(self, u, v, **kwargs):
"""Add directed edge ensuring acyclicity."""
def is_dag(self):
"""Check if graph is a valid DAG."""
def topological_sort(self):
"""Return topological ordering of nodes."""
def get_roots(self):
"""Get nodes with no parents."""
def get_leaves(self):
"""Get nodes with no children."""
class PDAG:
def __init__(self, ebunch=None):
"""Partially Directed Acyclic Graph."""
def add_edge(self, u, v, edge_type='directed'):
"""Add edge with specified type (directed/undirected)."""
class UndirectedGraph:
def __init__(self, ebunch=None):
"""Undirected graph base class."""
def add_edge(self, u, v, **kwargs):
"""Add undirected edge."""
def is_connected(self):
"""Check if graph is connected."""Markov Networks
Undirected graphical models representing joint probability distributions through clique potentials.
class MarkovNetwork:
def __init__(self, ebunch=None, latents=[]):
"""
Create a Markov Network (undirected graphical model).
Parameters:
- ebunch: input graph data (edge list or NetworkX graph)
- latents: list of latent variables
"""
def add_edge(self, u, v, **kwargs):
"""Add undirected edge between u and v."""
def add_factors(self, *factors):
"""Associate factors (clique potentials) to the graph."""
def get_factors(self, node=None):
"""Returns factors containing the node or all factors."""
def remove_factors(self, *factors):
"""Removes given factors from the model."""
def get_cardinality(self, node=None):
"""Returns cardinality of node or all variables."""
def check_model(self):
"""Check the model for various errors."""
def to_factor_graph(self):
"""Converts the Markov Model into Factor Graph."""
def triangulate(self, heuristic="H6", order=None, inplace=False):
"""Triangulate the graph."""
def to_junction_tree(self):
"""Creates a junction tree for the markov model."""
def markov_blanket(self, node):
"""Returns markov blanket for a random variable."""
def to_bayesian_model(self):
"""Creates a Bayesian Model which is minimum I-Map."""
def get_partition_function(self):
"""Returns partition function for the graph."""Factor Graphs
Bipartite graphs representing factorization of functions with variable and factor nodes.
class FactorGraph:
def __init__(self, ebunch=None):
"""
Create a Factor Graph (bipartite graph).
Parameters:
- ebunch: input graph data (edge list)
"""
def add_edge(self, u, v, **kwargs):
"""Add edge between variable_node and factor_node."""
def add_factors(self, *factors, replace=False):
"""Associate factors to the graph."""
def remove_factors(self, *factors):
"""Removes given factors."""
def get_variable_nodes(self):
"""Returns variable nodes in the graph."""
def get_factor_nodes(self):
"""Returns factor nodes in the graph."""
def to_markov_model(self):
"""Converts factor graph into markov model."""
def to_junction_tree(self):
"""Create junction tree for the factor graph."""
def get_partition_function(self):
"""Returns partition function."""Junction Trees and Cluster Graphs
Tree structures for efficient exact inference in probabilistic graphical models.
class JunctionTree:
def __init__(self, ebunch=None):
"""
Create a Junction Tree.
Parameters:
- ebunch: input graph data (edge list)
"""
def add_edge(self, u, v, **kwargs):
"""Add edge between two clique nodes."""
def check_model(self):
"""Check the model for various errors."""
class ClusterGraph:
def __init__(self, ebunch=None):
"""
Create a Cluster Graph.
Parameters:
- ebunch: input graph data (edge list)
"""
def add_node(self, node, **kwargs):
"""Add single node (clique) to the cluster graph."""
def add_nodes_from(self, nodes, **kwargs):
"""Add multiple nodes."""
def add_edge(self, u, v, **kwargs):
"""Add edge between two clique nodes."""
def add_factors(self, *factors):
"""Associate factors to the graph."""
def get_factors(self, node=None):
"""Return factors for node or all factors."""
def remove_factors(self, *factors):
"""Removes given factors."""
def get_partition_function(self):
"""Returns partition function."""Dynamic and Temporal Models
Models for time-series and temporal probabilistic relationships.
class DynamicBayesianNetwork:
def __init__(self, ebunch=None):
"""
Create a Dynamic Bayesian Network for temporal modeling.
Parameters:
- ebunch: input graph data (edge list)
"""
def add_edge(self, start, end, **kwargs):
"""Add edge between two nodes with time slices."""
def get_intra_edges(self, time_slice=0):
"""Returns intra slice edges."""
def get_inter_edges(self):
"""Returns inter-slice edges."""
def get_interface_nodes(self, time_slice=0):
"""Returns interface nodes."""
def get_slice_nodes(self, time_slice=0):
"""Returns nodes in particular timeslice."""
def add_cpds(self, *cpds):
"""Add CPDs to the network."""
def get_cpds(self, node=None, time_slice=None):
"""Returns CPDs for node/timeslice."""
def initialize_initial_state(self):
"""Re-adjust CPDs and edges."""
def get_constant_bn(self, t_slice=0):
"""Returns normal Bayesian Network object."""
def simulate(self, n_samples=10, n_time_slices=2, do=None, evidence=None):
"""Simulates time-series data."""
class MarkovChain:
def __init__(self, variables=None, card=None, start_state=None):
"""
Create a Markov Chain with multiple kernels.
Parameters:
- variables: list of variables of the model
- card: list of cardinalities of the variables
- start_state: list of tuples representing starting states
"""
def set_start_state(self, start_state):
"""Set the start state of the Markov Chain."""
def add_variable(self, variable, card=0):
"""Add a variable to the model."""
def add_variables_from(self, variables, cards):
"""Add several variables at once."""
def add_transition_model(self, variable, transition_model):
"""Adds transition model for variable."""
def sample(self, start_state=None, size=1, seed=None):
"""Sample from the Markov Chain."""
def generate_sample(self, start_state=None, size=1, seed=None):
"""Generator version of sample."""
def is_stationarity(self, tolerance=0.2, sample=None):
"""Check if chain is stationary."""Specialized Bayesian Networks
Specialized network types for specific use cases and variable types.
class NaiveBayes:
def __init__(self, feature_vars=None, dependent_var=None):
"""
Create a Naive Bayes model.
Parameters:
- feature_vars: list of predictor variables (features)
- dependent_var: dependent variable to be predicted
"""
def add_edge(self, u, v, *kwargs):
"""Add edge between dependent and feature variable."""
def add_edges_from(self, ebunch):
"""Add edges to the model."""
def fit(self, data, parent_node=None, estimator=None):
"""Computes CPDs from data."""
class LinearGaussianBayesianNetwork:
def __init__(self, ebunch=None, latents=set(), lavaan_str=None, dagitty_str=None):
"""
Create a Linear Gaussian Bayesian Network.
Parameters:
- ebunch: input graph data
- latents: set of latent variables
- lavaan_str: lavaan syntax string
- dagitty_str: dagitty syntax string
"""
def add_cpds(self, *cpds):
"""Add linear Gaussian CPDs to the network."""
def get_cpds(self, node=None):
"""Returns CPD of node or all CPDs."""
def remove_cpds(self, *cpds):
"""Removes given CPDs."""
def to_joint_gaussian(self):
"""Returns mean and covariance of joint gaussian distribution."""
def simulate(self, n=1000, seed=None):
"""Simulates data from the model."""
def fit(self, data, method="mle"):
"""Estimates parameters from data."""
def predict(self, data, distribution="joint"):
"""Predicts missing variables."""
class FunctionalBayesianNetwork:
def __init__(self, ebunch=None, latents=set(), lavaan_str=None, dagitty_str=None):
"""
Create a Functional Bayesian Network using Pyro distributions.
Parameters:
- ebunch: list of edges to build the network
- latents: set of latent variables
- lavaan_str: lavaan syntax string
- dagitty_str: dagitty syntax string
"""
def add_cpds(self, *cpds):
"""Adds FunctionalCPDs to the network."""
def get_cpds(self, node=None):
"""Returns CPD of node or all CPDs."""
def remove_cpds(self, *cpds):
"""Removes given CPDs."""
def simulate(self, n_samples=1000, seed=None):
"""Simulate samples from the model."""
def fit(self, data, method="SVI", optimizer=None, prior_fn=None):
"""Fit model using Pyro's SVI or MCMC."""Usage Examples
Creating a Simple Bayesian Network
from pgmpy.models import DiscreteBayesianNetwork
from pgmpy.factors.discrete import TabularCPD
# Create network structure
model = DiscreteBayesianNetwork([('Rain', 'Sprinkler'),
('Rain', 'Wet_Grass'),
('Sprinkler', 'Wet_Grass')])
# Define CPDs
cpd_rain = TabularCPD(variable='Rain', variable_card=2,
values=[[0.8], [0.2]])
cpd_sprinkler = TabularCPD(variable='Sprinkler', variable_card=2,
values=[[0.9, 0.1], [0.1, 0.9]],
evidence=['Rain'], evidence_card=[2])
# Add CPDs and validate
model.add_cpds(cpd_rain, cpd_sprinkler)
assert model.check_model()Working with Markov Networks
from pgmpy.models import MarkovNetwork
from pgmpy.factors.discrete import DiscreteFactor
# Create undirected model
model = MarkovNetwork([('A', 'B'), ('B', 'C'), ('C', 'A')])
# Add factors (clique potentials)
factor_ab = DiscreteFactor(['A', 'B'], [2, 2], [1, 2, 3, 4])
factor_bc = DiscreteFactor(['B', 'C'], [2, 2], [2, 1, 1, 2])
model.add_factors(factor_ab, factor_bc)
assert model.check_model()Install with Tessl CLI
npx tessl i tessl/pypi-pgmpy