tessl/pypi-pgmpy
A library for Probabilistic Graphical Models
—
Pending
factors.mddocs/
Probability Factors
Representations of probability distributions including discrete factors, conditional probability distributions, and continuous distributions. Factors are the fundamental building blocks for representing uncertainty in probabilistic graphical models.
Capabilities
Discrete Factors
Core discrete probability factor class for representing joint probability distributions over discrete variables.
class DiscreteFactor:
def __init__(self, variables, cardinality, values, state_names={}):
"""
Create a discrete factor over specified variables.
Parameters:
- variables: list of variable names
- cardinality: list of variable cardinalities (number of states)
- values: factor values as flat array or nested array
- state_names: dict mapping variables to their state names
"""
def scope(self):
"""
Get the variables in the factor's scope.
Returns:
list: Variable names in the factor
"""
def get_cardinality(self, variables):
"""Get cardinalities of specified variables."""
def get_value(self, **kwargs):
"""
Get factor value for specific variable assignments.
Parameters:
- kwargs: variable assignments as keyword arguments
Returns:
float: Factor value for the assignment
"""
def set_value(self, value, **kwargs):
"""Set factor value for specific variable assignments."""
def marginalize(self, variables, inplace=True):
"""
Sum out variables from the factor.
Parameters:
- variables: list of variables to marginalize out
- inplace: whether to modify factor in-place
Returns:
DiscreteFactor: Marginalized factor
"""
def maximize(self, variables, inplace=True):
"""
Maximize out variables (max-marginalization).
Parameters:
- variables: list of variables to maximize out
- inplace: whether to modify factor in-place
Returns:
DiscreteFactor: Maximized factor
"""
def normalize(self, inplace=True):
"""
Normalize factor values to sum to 1.
Parameters:
- inplace: whether to modify factor in-place
Returns:
DiscreteFactor: Normalized factor
"""
def reduce(self, values, inplace=True, show_warnings=True):
"""
Condition the factor on observed evidence.
Parameters:
- values: dict of variable assignments {variable: value}
- inplace: whether to modify factor in-place
- show_warnings: whether to show warnings
Returns:
DiscreteFactor: Reduced factor
"""
def sum(self, phi1, inplace=True):
"""Add another factor to this factor."""
def product(self, phi1, inplace=True):
"""
Multiply this factor with another factor.
Parameters:
- phi1: DiscreteFactor to multiply with
- inplace: whether to modify factor in-place
Returns:
DiscreteFactor: Product factor
"""
def divide(self, phi1, inplace=True):
"""Divide this factor by another factor."""
def sample(self, n, seed=None):
"""
Generate samples from the factor distribution.
Parameters:
- n: number of samples to generate
- seed: random seed for reproducibility
Returns:
pandas.DataFrame: Generated samples
"""
def copy(self):
"""Create deep copy of the factor."""Conditional Probability Distributions
Specialized factors representing conditional probability distributions P(X|Parents).
class TabularCPD:
def __init__(self, variable, variable_card, values, evidence=None,
evidence_card=None, state_names={}):
"""
Create a tabular conditional probability distribution.
Parameters:
- variable: name of the variable this CPD represents
- variable_card: cardinality of the variable
- values: CPD values as 2D array
- evidence: list of parent variable names
- evidence_card: list of parent variable cardinalities
- state_names: dict mapping variables to state names
"""
def get_values(self):
"""
Get the CPD values.
Returns:
numpy.ndarray: CPD probability values
"""
def normalize(self, inplace=True):
"""
Normalize CPD so each column sums to 1.
Parameters:
- inplace: whether to modify CPD in-place
Returns:
TabularCPD: Normalized CPD
"""
def marginalize(self, variables, inplace=True):
"""Marginalize over specified variables."""
def reduce(self, values, inplace=True, show_warnings=True):
"""
Condition CPD on observed evidence.
Parameters:
- values: dict of evidence {variable: value}
- inplace: whether to modify CPD in-place
- show_warnings: whether to show warnings
Returns:
TabularCPD: Reduced CPD
"""
def to_factor(self):
"""
Convert CPD to a DiscreteFactor.
Returns:
DiscreteFactor: Equivalent factor representation
"""
def reorder_parents(self, new_order, inplace=True):
"""Reorder parent variables to match specified order."""
def get_evidence(self):
"""
Get list of parent (evidence) variables.
Returns:
list: Parent variable names
"""
def copy(self):
"""Create deep copy of the CPD."""
def to_csv(self, filename):
"""Save CPD to CSV file."""
def to_dataframe(self):
"""
Convert CPD to pandas DataFrame.
Returns:
pandas.DataFrame: CPD as DataFrame
"""State Representation
Utility class for representing variable states.
class State:
def __init__(self, variable, state):
"""
Create a variable state.
Parameters:
- variable: variable name
- state: state value/index
"""
def __eq__(self, other):
"""Check equality with another State."""
def __hash__(self):
"""Hash function for use in sets/dictionaries."""Joint Probability Distributions
Full joint probability distributions over all variables.
class JointProbabilityDistribution:
def __init__(self, variables, cardinality, values):
"""
Create a joint probability distribution.
Parameters:
- variables: list of all variable names
- cardinality: list of variable cardinalities
- values: joint probability values
"""
def marginal_distribution(self, variables, inplace=True):
"""
Get marginal distribution over specified variables.
Parameters:
- variables: list of variables for marginal
- inplace: whether to modify distribution in-place
Returns:
JointProbabilityDistribution: Marginal distribution
"""
def conditional_distribution(self, variables, evidence):
"""
Get conditional distribution P(variables|evidence).
Parameters:
- variables: list of query variables
- evidence: dict of evidence {variable: value}
Returns:
JointProbabilityDistribution: Conditional distribution
"""
def check_normalization(self):
"""Check if distribution is properly normalized."""
def normalize(self, inplace=True):
"""Normalize distribution to sum to 1."""
def copy(self):
"""Create deep copy of the distribution."""Noisy-OR CPDs
Specialized CPDs implementing the noisy-OR model for causal relationships.
class NoisyORCPD:
def __init__(self, variable, evidence_card, leak_probability,
inhibition_probs):
"""
Create a Noisy-OR conditional probability distribution.
Parameters:
- variable: child variable name
- evidence_card: list of parent variable cardinalities
- leak_probability: probability of effect without any cause
- inhibition_probs: list of inhibition probabilities for each parent
"""
def to_tabular_cpd(self):
"""
Convert to equivalent TabularCPD.
Returns:
TabularCPD: Equivalent tabular representation
"""
def copy(self):
"""Create deep copy of the Noisy-OR CPD."""Continuous Factors
Linear Gaussian conditional probability distributions for continuous variables.
class LinearGaussianCPD:
def __init__(self, variable, beta, std, evidence=[]):
"""
Create a Linear Gaussian CPD for continuous variables.
Parameters:
- variable: name of the continuous variable
- beta: regression coefficients (including intercept)
- std: standard deviation of the noise
- evidence: list of parent variable names
"""
def copy(self):
"""Create deep copy of the Linear Gaussian CPD."""
@staticmethod
def get_random(variable, evidence, loc=0.0, scale=1.0, seed=None):
"""
Generate random Linear Gaussian CPD.
Parameters:
- variable: variable name
- evidence: list of parent variables
- loc: mean of coefficient distribution
- scale: standard deviation of coefficient distribution
- seed: random seed
Returns:
LinearGaussianCPD: Random Linear Gaussian CPD
"""Hybrid Factors
Functional CPDs that can represent arbitrary distributions using functional forms.
class FunctionalCPD:
def __init__(self, variable, evidence, distribution_fn, *args, **kwargs):
"""
Create a Functional CPD using arbitrary distribution functions.
Parameters:
- variable: name of the variable
- evidence: list of parent variable names
- distribution_fn: function returning pyro.distributions object
- args, kwargs: additional arguments for the distribution function
"""
def copy(self):
"""Create deep copy of the Functional CPD."""
def sample(self, evidence_dict=None, size=1, seed=None):
"""
Sample from the functional CPD given evidence.
Parameters:
- evidence_dict: dict of evidence values for parent variables
- size: number of samples to generate
- seed: random seed
Returns:
numpy.ndarray: Generated samples
"""Factor Operations
Utility functions for factor manipulation.
def factor_product(*factors):
"""
Compute product of multiple factors.
Parameters:
- factors: DiscreteFactor objects to multiply
Returns:
DiscreteFactor: Product of all factors
"""
def factor_divide(phi1, phi2):
"""
Divide one factor by another factor.
Parameters:
- phi1: numerator factor
- phi2: denominator factor
Returns:
DiscreteFactor: Division result
"""
def factor_sum_product(factors, variables):
"""
Compute sum-product of factors over specified variables.
Parameters:
- factors: list of DiscreteFactor objects
- variables: list of variables to sum out
Returns:
DiscreteFactor: Result of sum-product operation
"""Factor Collections
Classes for managing collections of factors.
class FactorSet:
def __init__(self, factors=None):
"""
Create a set of factors.
Parameters:
- factors: list of DiscreteFactor objects
"""
def product(self, inplace=True):
"""Compute product of all factors in the set."""
def marginalize(self, variables, inplace=True):
"""Marginalize variables from all factors."""
def reduce(self, values, inplace=True):
"""Reduce all factors with evidence."""
def factorset_product(*factor_sets):
"""Compute product of multiple factor sets."""
def factorset_divide(factor_set1, factor_set2):
"""Divide one factor set by another."""
class FactorDict:
def __init__(self, factors=None):
"""Dictionary-based factor collection with variable indexing."""
def get_factors(self, variables):
"""Get factors involving specified variables."""
def add_factors(self, *factors):
"""Add factors to the collection."""Usage Examples
Creating and Manipulating Discrete Factors
from pgmpy.factors.discrete import DiscreteFactor
# Create a factor over variables A and B
factor_ab = DiscreteFactor(['A', 'B'], [2, 2], [0.8, 0.2, 0.3, 0.7])
# Create another factor over variables B and C
factor_bc = DiscreteFactor(['B', 'C'], [2, 2], [0.9, 0.1, 0.4, 0.6])
# Multiply factors
result = factor_ab.product(factor_bc, inplace=False)
# Marginalize out variable B
marginal_ac = result.marginalize(['B'], inplace=False)
# Condition on evidence
evidence = {'A': 1}
conditional = result.reduce(evidence, inplace=False)Working with CPDs
from pgmpy.factors.discrete import TabularCPD
import numpy as np
# Create CPD for P(Grade | Difficulty, Intelligence)
values = np.array([[0.3, 0.4, 0.9, 0.08], # Grade = 0
[0.4, 0.25, 0.08, 0.02], # Grade = 1
[0.3, 0.35, 0.02, 0.9]]) # Grade = 2
cpd = TabularCPD(variable='Grade', variable_card=3, values=values,
evidence=['Difficulty', 'Intelligence'],
evidence_card=[2, 2],
state_names={'Grade': ['A', 'B', 'C'],
'Difficulty': ['Easy', 'Hard'],
'Intelligence': ['Low', 'High']})
# Check if CPD is normalized
print(cpd.get_values().sum(axis=0)) # Should sum to 1 for each column
# Convert to factor
factor = cpd.to_factor()
# Export to CSV
cpd.to_csv('grade_cpd.csv')Linear Gaussian CPDs for Continuous Variables
from pgmpy.factors.continuous import LinearGaussianCPD
# Create P(Y | X) = N(2 + 1.5*X, σ=0.5)
cpd_y = LinearGaussianCPD(variable='Y',
beta=[2, 1.5], # intercept + coefficient
std=0.5,
evidence=['X'])
# Generate random Linear Gaussian CPD
random_cpd = LinearGaussianCPD.get_random('Z', ['X', 'Y'], seed=42)Install with Tessl CLI
npx tessl i tessl/pypi-pgmpy