Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
28 changes: 27 additions & 1 deletion pycid/core/causal_bayesian_network.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,7 @@
import networkx as nx
import numpy as np
from pgmpy.factors.discrete import TabularCPD
from pgmpy.inference.ExactInference import BeliefPropagation
from pgmpy.inference.ExactInference import BeliefPropagation, VariableElimination
from pgmpy.models import BayesianNetwork
from pgmpy.sampling import BayesianModelSampling

Expand Down Expand Up @@ -194,11 +194,29 @@ def intervene(self, intervention: Dict[str, Outcome]) -> None:
new_dist[intervention[variable]] = 1
self.model[variable] = new_dist

def _context_has_zero_probability(
self, context: Dict[str, Outcome], intervention: Optional[Dict[str, Outcome]] = None
) -> bool:
"""Return whether the conditioning context has probability zero."""
if not context:
return False

cbn = self.copy()
if intervention:
cbn.intervene(intervention)

# Test event probability without conditioning on the event. Decision
# domains still provide valid CPDs for this diagnostic check.
factor = VariableElimination(cbn).query(list(context), show_progress=False)
index = tuple(factor.state_names[variable].index(context[variable]) for variable in factor.variables)
return bool(factor.values[index] == 0)

def expected_value(
self,
variables: Iterable[str],
context: Dict[str, Outcome],
intervention: Optional[Dict[str, Outcome]] = None,
zero_prob_value: Optional[float] = None,
) -> List[float]:
"""Compute the expected value of a real-valued variable for a given context,
under an optional intervention
Expand All @@ -210,8 +228,16 @@ def expected_value(
context: Node values to condition upon. A dictionary mapping of node => value.

intervention: Interventions to apply. A dictionary mapping node => value.

zero_prob_value: Value to return for each variable if context has zero probability.
"""
factor = self.query(variables, context, intervention=intervention)
# Conditional expectations on zero-probability contexts are undefined.
# Confirm that cause directly, instead of treating every NaN as this case.
if np.isnan(factor.values).all() and self._context_has_zero_probability(context, intervention):
if zero_prob_value is not None:
return [zero_prob_value for _ in factor.variables]
raise RuntimeError(f"query {variables} | {context} conditions on a zero-probability event")
factor.normalize() # make probs add to one

ev = np.array([0.0 for _ in factor.variables])
Expand Down
4 changes: 3 additions & 1 deletion pycid/core/macid_base.py
Original file line number Diff line number Diff line change
Expand Up @@ -394,7 +394,9 @@ def opt_policy(**parent_values: Outcome) -> Outcome:
eu = {}
for d in domain:
parent_values[decision] = d
eu[d] = sum(copy.expected_value(descendant_utility_nodes, parent_values))
# Unreachable decision contexts do not affect expected utility, so
# use a neutral fallback while constructing the decision rule.
eu[d] = sum(copy.expected_value(descendant_utility_nodes, parent_values, zero_prob_value=0.0))
return max(eu, key=eu.get) # type: ignore

self.add_cpds(StochasticFunctionCPD(decision, opt_policy, self, domain=domain, label="opt"))
Expand Down