API Reference

This page is auto-generated from source docstrings.

PC Algorithm

Bases: LearnAlgo

PC algorithm with stable variant (Colombo & Maathuis 2014).

Implements three v-structure determination rules from Ramsey et al. (2016): - Conservative: Orient as v-structure only if unanimous across separating sets - Majority: Orient if majority of separating sets do not contain the middle node - PC-Max: Orient based on highest p-value for independence

Source code in mixpc/pc_algorithm.py

class PC(LearnAlgo):
    """PC algorithm with stable variant (Colombo & Maathuis 2014).

    Implements three v-structure determination rules from Ramsey et al. (2016):
    - Conservative: Orient as v-structure only if unanimous across separating sets
    - Majority: Orient if majority of separating sets do not contain the middle node
    - PC-Max: Orient based on highest p-value for independence
    """

    def __init__(self, alpha: float = 0.05, test: type[CItest] = MixedFisherZ) -> None:
        """Initialize PC algorithm.

        Args:
            alpha (float, optional): Significance threshold for independence tests.
                Smaller values result in sparser graphs. Defaults to 0.05.
            test (type[CItest], optional): Conditional independence test class.
                Defaults to MixedFisherZ.
        """
        self.alpha: float = alpha
        self.pdag: PDAG = PDAG()
        self.skel: pd.DataFrame | None = None
        self.sep_sets: dict[tuple[str, str], set[str]] = {}
        self.ci_test = test()
        self.prior: PriorKnowledge | None = None
        self.ci_test_count: int = 0

    def learn_graph(
        self,
        data_dict: dict[str, np.ndarray],
        v_structure_rule: Literal["conservative", "majority", "pc-max"] = "conservative",
        prior_knowledge: PriorKnowledge | None = None,
    ) -> PDAG:
        """Learn causal graph using PC stable algorithm.

        Args:
            data_dict (dict[str, np.ndarray]): Dictionary mapping variable names to data arrays.
            v_structure_rule (Literal["conservative", "majority", "pc-max"], optional):
                v-structure determination rule from Ramsey et al. (2016).
                Defaults to "conservative".
            prior_knowledge (PriorKnowledge, optional): Edge/direction/layering hints
                consulted across all three phases. Defaults to None.

        Returns:
            PDAG: Partially directed acyclic graph.

        Raises:
            ValueError: If v_structure_rule is not recognized.
        """
        if v_structure_rule not in {"conservative", "majority", "pc-max"}:
            raise ValueError(
                f"v_structure_rule must be 'conservative', 'majority', or 'pc-max', got {v_structure_rule}"
            )

        if prior_knowledge is not None:
            prior_knowledge.validate(set(data_dict.keys()))
        self.prior = prior_knowledge
        self.ci_test_count = 0

        # Phase 1: Skeleton learning (PC stable, optionally constrained by prior knowledge)
        self._find_skeleton_stable(data=data_dict, alpha=self.alpha)

        # Pre-orient edges with a uniquely allowed direction (layering / required_directions /
        # forbidden_directions that pin the alternative). Done before v-structure phase so that
        # downstream rules see the constraints as already-decided orientations.
        if self.prior is not None:
            self._apply_prior_orientations()

        # Phase 2: V-structure orientation using chosen rule
        self._orient_v_structures(data=data_dict, alpha=self.alpha, rule=v_structure_rule)

        # Phase 3: Meek rule application for remaining undirected edges
        self._apply_meek_rules()

        # Final pass: if Meek introduced an orientation that prior knowledge forbids, flip it
        # when the reverse is allowed. Both-directions-forbidden edges are left as-is.
        if self.prior is not None:
            self._reconcile_with_prior()

        return self.pdag

    def _find_skeleton_stable(
        self,
        data: dict[str, np.ndarray],
        alpha: float = 0.05,
    ) -> None:
        """Find skeleton using PC stable algorithm (Colombo & Maathuis 2014).

        The stable version ensures deterministic skeleton discovery independent
        of test order.

        Args:
            data (dict[str, np.ndarray]): Dictionary mapping variable names to data arrays.
            alpha (float): Significance threshold for independence tests.
        """
        node_names = sorted(list(data.keys()))
        n_features = len(node_names)

        # Initialize complete undirected graph
        skeleton = pd.DataFrame(
            np.ones((n_features, n_features)) - np.eye(n_features),
            columns=node_names,
            index=node_names,
        )

        # Drop forbidden edges before the first CI test — they are never even considered.
        if self.prior is not None:
            for a, b in self.prior.forbidden_edges:
                skeleton.loc[a, b] = skeleton.loc[b, a] = 0

        self.sep_sets = {}
        d = 0

        # Iterate over conditioning set sizes until no pair can be tested further.
        while True:
            adj_pairs = self._get_adjacent_pairs(skeleton)
            if not adj_pairs:
                break

            # Snapshot adjacencies once per level (PC-stable: neighbors must not change mid-level)
            skeleton_snapshot = skeleton.copy()

            any_test_possible = False
            for i, j in adj_pairs:
                if self._try_separate_pair(
                    i, j, d=d, data=data, alpha=alpha,
                    skeleton=skeleton, skeleton_snapshot=skeleton_snapshot,
                ):
                    any_test_possible = True

            if not any_test_possible:
                break
            d += 1

        self.skel = skeleton
        self.pdag = PDAG.from_pandas_adjacency(skeleton)

    def _try_separate_pair(
        self,
        i: str,
        j: str,
        *,
        d: int,
        data: dict[str, np.ndarray],
        alpha: float,
        skeleton: pd.DataFrame,
        skeleton_snapshot: pd.DataFrame,
    ) -> bool:
        """Test the (i, j) pair at conditioning-set size ``d``; mutates ``skeleton`` if separated.

        Returns ``True`` when a CI test was actually attempted at this level (so the caller
        knows to keep iterating with a larger ``d``).
        """
        if skeleton.loc[i, j] == 0:
            return False
        if self.prior is not None and self.prior.is_required_edge(i, j):
            return False

        candidate_neighbors = self._candidate_separators(i, j, skeleton_snapshot)
        if self.prior is not None:
            candidate_neighbors = set(
                self.prior.filter_separating_set(i, j, list(candidate_neighbors))
            )
        if len(candidate_neighbors) < d:
            return False

        for sep_set_subset in combinations(sorted(candidate_neighbors), d):
            sep_set_list = list(sep_set_subset)
            if not sep_set_list:
                _, p_value = self.ci_test.test(x_data=data[i], y_data=data[j])
            else:
                z_data = np.concatenate([data[node] for node in sep_set_list], axis=1)
                _, p_value = self.ci_test.test(x_data=data[i], y_data=data[j], z_data=z_data)
            self.ci_test_count += 1

            if p_value >= alpha:
                skeleton.loc[i, j] = skeleton.loc[j, i] = 0
                self.sep_sets[(i, j)] = set(sep_set_list)
                self.sep_sets[(j, i)] = set(sep_set_list)
                return True
        return True

    def _candidate_separators(
        self, i: str, j: str, skeleton_snapshot: pd.DataFrame
    ) -> set[str]:
        """Union of neighbors of ``i`` and ``j`` (excluding each other) in the snapshot."""
        cols = skeleton_snapshot.columns
        row_i = skeleton_snapshot.loc[i].to_numpy()
        row_j = skeleton_snapshot.loc[j].to_numpy()
        neighbors_i = {str(cols[idx]) for idx, val in enumerate(row_i) if val == 1 and cols[idx] != j}
        neighbors_j = {str(cols[idx]) for idx, val in enumerate(row_j) if val == 1 and cols[idx] != i}
        return neighbors_i | neighbors_j

    def _get_adjacent_pairs(self, skeleton: pd.DataFrame) -> list[tuple[str, str]]:
        """Get all adjacent pairs in the skeleton.

        Args:
            skeleton (pd.DataFrame): Adjacency matrix of skeleton.

        Returns:
            list[tuple[str, str]]: List of adjacent node pairs (sorted).
        """
        pairs = []
        nodes = sorted(skeleton.columns.tolist())
        for i in range(len(nodes)):
            for j in range(i + 1, len(nodes)):
                if skeleton.iloc[i, j] == 1:
                    pairs.append((nodes[i], nodes[j]))
        return pairs

    def _find_unshielded_triples(self, pdag: PDAG) -> list[tuple[str, str, str]]:
        """Find all unshielded triples in the graph.

        An unshielded triple is (i, j, k) where i-j-k is a path and i,k are not adjacent.

        Args:
            pdag (PDAG): Current partially directed graph.

        Returns:
            list[tuple[str, str, str]]: List of unshielded triples (i, j, k).
        """
        triples = []
        for j in pdag.nodes:
            neighbors_j = pdag.undir_neighbors(j)
            if len(neighbors_j) < 2:
                continue

            # Find all pairs of neighbors
            for i, k in combinations(sorted(neighbors_j), 2):
                # Check if i and k are not adjacent (unshielded)
                if not pdag.is_adjacent(i, k):
                    triples.append((i, j, k))
        return triples

    def _orient_v_structures(
        self,
        data: dict[str, np.ndarray],
        alpha: float = 0.05,
        rule: Literal["conservative", "majority", "pc-max"] = "conservative",
    ) -> None:
        """Determine and orient v-structures using specified rule.

        Implements three rules from Ramsey et al. (2016) for v-structure discovery.

        Args:
            data (dict[str, np.ndarray]): Dictionary mapping variable names to data arrays.
            alpha (float): Significance threshold.
            rule (Literal["conservative", "majority", "pc-max"]): v-structure rule.
        """
        pdag = self.pdag.copy()
        unshielded_triples = self._find_unshielded_triples(pdag)

        for i, j, k in unshielded_triples:
            # Get all potential separating sets for (i, k)
            potential_sep_sets = self._get_potential_separating_sets(i, k, pdag, data)

            # Determine if this is a v-structure based on chosen rule.
            if rule == "conservative":
                is_v_structure = self._conservative_v_structure_rule(i, j, k, potential_sep_sets, alpha)
            elif rule == "majority":
                is_v_structure = self._majority_v_structure_rule(i, j, k, potential_sep_sets, alpha)
            elif rule == "pc-max":
                is_v_structure = self._pc_max_v_structure_rule(i, j, k, potential_sep_sets, alpha)
            else:
                raise ValueError(f"Unknown rule: {rule}")

            if is_v_structure:
                # Skip when the proposed v-structure conflicts with prior knowledge: orienting
                # i -> j (or k -> j) is forbidden — for example, j is in an earlier layer.
                if self.prior is not None and (
                    self.prior.is_forbidden_direction(i, j) or self.prior.is_forbidden_direction(k, j)
                ):
                    continue

                # Orient as v-structure: i -> j <- k (only if edges are still undirected)
                try:
                    if pdag.is_adjacent(i, j) and ((i, j) in pdag.undir_edges or (j, i) in pdag.undir_edges):
                        pdag.undir_to_dir_edge(tail=i, head=j)
                    if pdag.is_adjacent(k, j) and ((k, j) in pdag.undir_edges or (j, k) in pdag.undir_edges):
                        pdag.undir_to_dir_edge(tail=k, head=j)
                except AssertionError:
                    # Edge may have been already oriented - skip
                    pass

        self.pdag = pdag

    def _get_potential_separating_sets(
        self, i: str, k: str, pdag: PDAG, data: dict[str, np.ndarray]
    ) -> list[tuple[set[str], float]]:
        """Get all potential separating sets for a pair of nodes with p-values.

        Args:
            i (str): First node.
            k (str): Second node.
            pdag (PDAG): Current graph.
            data (dict[str, np.ndarray]): Dictionary mapping variable names to data arrays.

        Returns:
            list[tuple[set[str], float]]: List of (separating set, p-value) tuples.
        """
        potential_sets = []

        # Get neighbors of both nodes
        neighbors_i = pdag.neighbors(i)
        neighbors_k = pdag.neighbors(k)

        # Get all combinations of neighbors (excluding i and k)
        all_neighbors = neighbors_i.union(neighbors_k)
        all_neighbors.discard(i)
        all_neighbors.discard(k)

        # Layering: drop "future" nodes from the conditioning pool to keep v-structure
        # decisions consistent with the skeleton phase.
        if self.prior is not None:
            all_neighbors = set(self.prior.filter_separating_set(i, k, list(all_neighbors)))

        # Test all possible subsets
        for r in range(len(all_neighbors) + 1):
            for sep_set in combinations(sorted(all_neighbors), r):
                sep_set_list = list(sep_set)

                # Perform independence test
                if not sep_set_list:
                    _, p_value = self.ci_test.test(x_data=data[i], y_data=data[k])
                else:
                    z_data = np.concatenate(
                        [data[node] for node in sep_set_list],
                        axis=1,
                    )
                    _, p_value = self.ci_test.test(
                        x_data=data[i],
                        y_data=data[k],
                        z_data=z_data,
                    )
                self.ci_test_count += 1

                potential_sets.append((set(sep_set), p_value))

        return potential_sets

    def _conservative_v_structure_rule(
        self,
        i: str,
        j: str,
        k: str,
        potential_sep_sets: list[tuple[set[str], float]],
        alpha: float,
    ) -> bool:
        """Conservative rule: Orient as v-structure only if unanimous.

        A v-structure is oriented only if all separating sets that make (i,k)
        independent do NOT contain j.

        Args:
            i (str): First parent.
            j (str): Middle node.
            k (str): Second parent.
            potential_sep_sets (list[tuple[set[str], float]]): Potential separating sets.
            alpha (float): Significance threshold.

        Returns:
            bool: True if v-structure should be oriented.
        """
        # Find all separating sets that render i and k independent
        independent_sep_sets = [sep_set for sep_set, p_value in potential_sep_sets if p_value >= alpha]

        if not independent_sep_sets:
            # If no separating set makes i,k independent, it's a v-structure
            return True

        # Orient only when every separating set excludes the middle node.
        return all(j not in sep_set for sep_set in independent_sep_sets)

    def _majority_v_structure_rule(
        self,
        i: str,
        j: str,
        k: str,
        potential_sep_sets: list[tuple[set[str], float]],
        alpha: float,
    ) -> bool:
        """Majority rule: Orient if majority of separating sets do not contain j.

        Args:
            i (str): First parent.
            j (str): Middle node.
            k (str): Second parent.
            potential_sep_sets (list[tuple[set[str], float]]): Potential separating sets.
            alpha (float): Significance threshold.

        Returns:
            bool: True if v-structure should be oriented.
        """
        # Find all separating sets that render i and k independent
        independent_sep_sets = [sep_set for sep_set, p_value in potential_sep_sets if p_value >= alpha]

        if not independent_sep_sets:
            # If no separating set makes i,k independent, it's a v-structure
            return True

        # Count how many separating sets do NOT contain j
        count_without_j = sum(1 for sep_set in independent_sep_sets if j not in sep_set)

        # Orient as v-structure if majority do not contain j
        return count_without_j > len(independent_sep_sets) / 2

    def _pc_max_v_structure_rule(
        self,
        i: str,
        j: str,
        k: str,
        potential_sep_sets: list[tuple[set[str], float]],
        alpha: float,
    ) -> bool:
        """PC-Max rule: Orient based on highest p-value for independence.

        Compares the highest p-value obtained when conditioning on sets excluding j
        with the highest p-value when including sets with j. Orients as v-structure
        if independence is more likely with j excluded.

        Args:
            i (str): First parent.
            j (str): Middle node.
            k (str): Second parent.
            potential_sep_sets (list[tuple[set[str], float]]): Potential separating sets.
            alpha (float): Significance threshold (not directly used in this rule).

        Returns:
            bool: True if v-structure should be oriented.
        """
        # Separate p-values by whether j is in the separating set
        p_values_without_j = [p for sep_set, p in potential_sep_sets if j not in sep_set]
        p_values_with_j = [p for sep_set, p in potential_sep_sets if j in sep_set]

        # Get maximum p-values from each group
        max_p_without_j = max(p_values_without_j) if p_values_without_j else 0
        max_p_with_j = max(p_values_with_j) if p_values_with_j else 0

        # Orient as v-structure if independence is more likely without j
        return max_p_without_j > max_p_with_j

    def _apply_prior_orientations(self) -> None:
        """Orient every undirected edge whose direction is uniquely fixed by prior knowledge.

        Runs after skeleton discovery and before v-structure orientation. Layering between
        layers, explicit ``required_directions``, and ``forbidden_directions`` that pin the
        alternative all flow through ``PriorKnowledge.required_direction_for``.
        """
        assert self.prior is not None
        pdag = self.pdag.copy()
        for i, j in list(pdag.undir_edges):
            forced = self.prior.required_direction_for(i, j)
            if forced is None:
                continue
            tail, head = forced
            # Edge may have been oriented by an earlier iteration of this loop — ignore.
            with suppress(AssertionError):
                pdag.undir_to_dir_edge(tail=tail, head=head)
        self.pdag = pdag

    def _reconcile_with_prior(self) -> None:
        """Flip directed edges Meek introduced in a forbidden direction when the reverse is allowed.

        If both directions are forbidden, the edge is left untouched: the user contradicted
        themselves about an edge that PC nonetheless found in the skeleton, and silently
        rewriting it is worse than surfacing the inconsistency.
        """
        assert self.prior is not None
        pdag = self.pdag.copy()
        flipped = False
        for tail, head in list(pdag.dir_edges):
            if not self.prior.is_forbidden_direction(tail, head):
                continue
            if self.prior.is_forbidden_direction(head, tail):
                continue
            # Use the public remove + private add to swap orientation in place.
            pdag.remove_edge(tail, head)
            pdag._add_dir_edge(head, tail)
            flipped = True
        if flipped:
            self.pdag = pdag

    def _apply_meek_rules(self) -> None:
        """Apply Meek rules to orient remaining undirected edges.

        Implements rules R1-R4 from Meek (1995) to maximize the number of
        directed edges consistent with acyclicity.
        """
        pdag = self.pdag.copy()

        # Apply rules until convergence to maximize orientations.
        while True:
            before = (set(pdag.undir_edges), set(pdag.dir_edges))
            pdag = rule_1(pdag=pdag)
            pdag = rule_2(pdag=pdag)
            pdag = rule_3(pdag=pdag)
            pdag = rule_4(pdag=pdag)
            after = (set(pdag.undir_edges), set(pdag.dir_edges))
            if after == before:
                break

        self.pdag = pdag

    @property
    def skeleton(self) -> pd.DataFrame:
        """Return the underlying skeleton as adjacency matrix.

        Returns:
            pd.DataFrame: Adjacency matrix of the skeleton.

        Raises:
            ValueError: If skeleton has not been learned yet.
        """
        if self.skel is None:
            raise ValueError("Skeleton not learned yet.")
        return self.skel

    @property
    def adjacency_matrix(self) -> pd.DataFrame:
        """Return the learned PDAG as adjacency matrix.

        Returns:
            pd.DataFrame: Adjacency matrix of the PDAG.
                - A[i,j]=1, A[j,i]=0: directed edge i→j
                - A[i,j]=1, A[j,i]=1: undirected edge i—j
                - A[i,j]=0, A[j,i]=0: no edge
        """
        return self.pdag.adjacency_matrix

    @property
    def causal_order(self) -> list[str] | None:
        """Return causal order if PDAG is fully directed (DAG).

        Returns:
            list[str] | None: Causal order if PDAG is a DAG, None otherwise.
        """
        if self.pdag.num_undir_edges == 0:
            dag = DAG(nodes=self.pdag.nodes, edges=self.pdag.dir_edges)
            return dag.causal_order
        return None

adjacency_matrix property

Return the learned PDAG as adjacency matrix.

Returns:

Type	Description
DataFrame	pd.DataFrame: Adjacency matrix of the PDAG. - A[i,j]=1, A[j,i]=0: directed edge i→j - A[i,j]=1, A[j,i]=1: undirected edge i—j - A[i,j]=0, A[j,i]=0: no edge

causal_order property

Return causal order if PDAG is fully directed (DAG).

Returns:

Type	Description
list[str] | None	list[str] | None: Causal order if PDAG is a DAG, None otherwise.

skeleton property

Return the underlying skeleton as adjacency matrix.

Returns:

Type	Description
DataFrame	pd.DataFrame: Adjacency matrix of the skeleton.

Raises:

Type	Description
ValueError	If skeleton has not been learned yet.

__init__(alpha=0.05, test=MixedFisherZ)

Initialize PC algorithm.

Parameters:

Name	Type	Description	Default
alpha	float	Significance threshold for independence tests. Smaller values result in sparser graphs. Defaults to 0.05.	0.05
test	type[CItest]	Conditional independence test class. Defaults to MixedFisherZ.	MixedFisherZ

Source code in mixpc/pc_algorithm.py

def __init__(self, alpha: float = 0.05, test: type[CItest] = MixedFisherZ) -> None:
    """Initialize PC algorithm.

    Args:
        alpha (float, optional): Significance threshold for independence tests.
            Smaller values result in sparser graphs. Defaults to 0.05.
        test (type[CItest], optional): Conditional independence test class.
            Defaults to MixedFisherZ.
    """
    self.alpha: float = alpha
    self.pdag: PDAG = PDAG()
    self.skel: pd.DataFrame | None = None
    self.sep_sets: dict[tuple[str, str], set[str]] = {}
    self.ci_test = test()
    self.prior: PriorKnowledge | None = None
    self.ci_test_count: int = 0

learn_graph(data_dict, v_structure_rule='conservative', prior_knowledge=None)

Learn causal graph using PC stable algorithm.

Parameters:

Name	Type	Description	Default
data_dict	dict[str, ndarray]	Dictionary mapping variable names to data arrays.	required
v_structure_rule	Literal['conservative', 'majority', 'pc-max']	v-structure determination rule from Ramsey et al. (2016). Defaults to "conservative".	'conservative'
prior_knowledge	PriorKnowledge	Edge/direction/layering hints consulted across all three phases. Defaults to None.	None

Returns:

Name	Type	Description
PDAG	PDAG	Partially directed acyclic graph.

Raises:

Type	Description
ValueError	If v_structure_rule is not recognized.

Source code in mixpc/pc_algorithm.py

def learn_graph(
    self,
    data_dict: dict[str, np.ndarray],
    v_structure_rule: Literal["conservative", "majority", "pc-max"] = "conservative",
    prior_knowledge: PriorKnowledge | None = None,
) -> PDAG:
    """Learn causal graph using PC stable algorithm.

    Args:
        data_dict (dict[str, np.ndarray]): Dictionary mapping variable names to data arrays.
        v_structure_rule (Literal["conservative", "majority", "pc-max"], optional):
            v-structure determination rule from Ramsey et al. (2016).
            Defaults to "conservative".
        prior_knowledge (PriorKnowledge, optional): Edge/direction/layering hints
            consulted across all three phases. Defaults to None.

    Returns:
        PDAG: Partially directed acyclic graph.

    Raises:
        ValueError: If v_structure_rule is not recognized.
    """
    if v_structure_rule not in {"conservative", "majority", "pc-max"}:
        raise ValueError(
            f"v_structure_rule must be 'conservative', 'majority', or 'pc-max', got {v_structure_rule}"
        )

    if prior_knowledge is not None:
        prior_knowledge.validate(set(data_dict.keys()))
    self.prior = prior_knowledge
    self.ci_test_count = 0

    # Phase 1: Skeleton learning (PC stable, optionally constrained by prior knowledge)
    self._find_skeleton_stable(data=data_dict, alpha=self.alpha)

    # Pre-orient edges with a uniquely allowed direction (layering / required_directions /
    # forbidden_directions that pin the alternative). Done before v-structure phase so that
    # downstream rules see the constraints as already-decided orientations.
    if self.prior is not None:
        self._apply_prior_orientations()

    # Phase 2: V-structure orientation using chosen rule
    self._orient_v_structures(data=data_dict, alpha=self.alpha, rule=v_structure_rule)

    # Phase 3: Meek rule application for remaining undirected edges
    self._apply_meek_rules()

    # Final pass: if Meek introduced an orientation that prior knowledge forbids, flip it
    # when the reverse is allowed. Both-directions-forbidden edges are left as-is.
    if self.prior is not None:
        self._reconcile_with_prior()

    return self.pdag

Prior Knowledge

User-supplied constraints consumed by :class:mixpc.pc_algorithm.PC.

All edge tuples are (tail, head). For undirected hints (required_edges, forbidden_edges) the ordering does not matter — both (a, b) and (b, a) are treated identically. For directed hints (required_directions, forbidden_directions) the tuple is read as tail -> head.

Parameters:

Name	Type	Description	Default
required_edges	list[Edge]	Edges that must appear in the skeleton (undirected sense). Skipped during CI testing so they are never removed.	list()
forbidden_edges	list[Edge]	Edges that must not appear in the skeleton. Removed from the initial complete graph; never tested.	list()
required_directions	list[Edge]	Edges pinned to a specific orientation tail -> head. Implies the edge is in the skeleton.	list()
forbidden_directions	list[Edge]	Orientations tail -> head that must never appear. The undirected edge may still exist.	list()
layering	list[list[str]] | None	Partial temporal order. layering[k] is the set of nodes in stage k; every node in stage k precedes every node in stage k+1. Order within a stage is unknown.	None

Source code in mixpc/prior_knowledge.py

@dataclass
class PriorKnowledge:
    """User-supplied constraints consumed by :class:`mixpc.pc_algorithm.PC`.

    All edge tuples are ``(tail, head)``. For undirected hints (``required_edges``,
    ``forbidden_edges``) the ordering does not matter — both ``(a, b)`` and
    ``(b, a)`` are treated identically. For directed hints
    (``required_directions``, ``forbidden_directions``) the tuple is read as
    ``tail -> head``.

    Args:
        required_edges: Edges that must appear in the skeleton (undirected
            sense). Skipped during CI testing so they are never removed.
        forbidden_edges: Edges that must not appear in the skeleton. Removed
            from the initial complete graph; never tested.
        required_directions: Edges pinned to a specific orientation
            ``tail -> head``. Implies the edge is in the skeleton.
        forbidden_directions: Orientations ``tail -> head`` that must never
            appear. The undirected edge may still exist.
        layering: Partial temporal order. ``layering[k]`` is the set of nodes
            in stage ``k``; every node in stage ``k`` precedes every node in
            stage ``k+1``. Order within a stage is unknown.
    """

    required_edges: list[Edge] = field(default_factory=list)
    forbidden_edges: list[Edge] = field(default_factory=list)
    required_directions: list[Edge] = field(default_factory=list)
    forbidden_directions: list[Edge] = field(default_factory=list)
    layering: list[list[str]] | None = None

    def __post_init__(self) -> None:
        """Init dataclass."""
        self._required_edges_set: set[frozenset[str]] = {frozenset(e) for e in self.required_edges}
        self._forbidden_edges_set: set[frozenset[str]] = {frozenset(e) for e in self.forbidden_edges}
        self._required_dir_set: set[Edge] = {(e[0], e[1]) for e in self.required_directions}
        self._forbidden_dir_set: set[Edge] = {(e[0], e[1]) for e in self.forbidden_directions}
        self._layer_of: dict[str, int] = {}
        if self.layering is not None:
            for idx, stage in enumerate(self.layering):
                for node in stage:
                    self._layer_of[node] = idx

    def validate(self, nodes: set[str]) -> None:
        """Check internal consistency and that every named node exists in ``nodes``."""
        self._validate_node_membership(nodes)
        self._validate_edge_conflicts()
        if self.layering is not None:
            self._validate_layering()

    def _validate_node_membership(self, nodes: set[str]) -> None:
        all_named = (
            {n for e in self.required_edges for n in e}
            | {n for e in self.forbidden_edges for n in e}
            | {n for e in self.required_directions for n in e}
            | {n for e in self.forbidden_directions for n in e}
            | set(self._layer_of.keys())
        )
        unknown = all_named - nodes
        if unknown:
            raise ValueError(f"Prior knowledge references unknown nodes: {sorted(unknown)}")
        all_edges = (
            self.required_edges + self.forbidden_edges + self.required_directions + self.forbidden_directions
        )
        for e in all_edges:
            if e[0] == e[1]:
                raise ValueError(f"Self-loop in prior knowledge: {e}")

    def _validate_edge_conflicts(self) -> None:
        overlap = self._required_edges_set & self._forbidden_edges_set
        if overlap:
            raise ValueError(f"Edges appear in both required_edges and forbidden_edges: {[tuple(e) for e in overlap]}")
        for tail, head in self._required_dir_set:
            if frozenset((tail, head)) in self._forbidden_edges_set:
                raise ValueError(f"required_direction {(tail, head)} contradicts a forbidden_edge.")
            if (tail, head) in self._forbidden_dir_set:
                raise ValueError(f"required_direction {(tail, head)} contradicts a forbidden_direction.")
            if (head, tail) in self._required_dir_set:
                raise ValueError(
                    f"Conflicting required_directions for the same edge: {(tail, head)} and {(head, tail)}."
                )

    def _validate_layering(self) -> None:
        assert self.layering is not None
        seen: set[str] = set()
        for stage in self.layering:
            dup = seen & set(stage)
            if dup:
                raise ValueError(f"Node(s) appear in multiple layers: {sorted(dup)}")
            seen |= set(stage)
        for tail, head in self._required_dir_set:
            if (
                tail in self._layer_of
                and head in self._layer_of
                and self._layer_of[tail] > self._layer_of[head]
            ):
                raise ValueError(
                    f"required_direction {tail} -> {head} contradicts layering "
                    f"(layer {self._layer_of[tail]} > layer {self._layer_of[head]})."
                )

    # ----- predicates used by PC ------------------------------------------------

    def is_forbidden_edge(self, i: str, j: str) -> bool:
        """Whether the undirected edge ``{i, j}`` is blacklisted."""
        return frozenset((i, j)) in self._forbidden_edges_set

    def is_required_edge(self, i: str, j: str) -> bool:
        """Whether the undirected edge ``{i, j}`` must appear (directly or via a required direction)."""
        if frozenset((i, j)) in self._required_edges_set:
            return True
        return (i, j) in self._required_dir_set or (j, i) in self._required_dir_set

    def is_forbidden_direction(self, tail: str, head: str) -> bool:
        """Whether the orientation ``tail -> head`` is forbidden by any hint or by layering."""
        if (tail, head) in self._forbidden_dir_set:
            return True
        if (head, tail) in self._required_dir_set:
            return True
        return (
            self.layering is not None
            and tail in self._layer_of
            and head in self._layer_of
            and self._layer_of[tail] > self._layer_of[head]
        )

    def required_direction_for(self, i: str, j: str) -> Edge | None:
        """Return the uniquely allowed orientation of edge {i, j}, if any.

        Resolution order: explicit required_direction → layering → forbidden_direction
        leaving exactly one valid side. Returns ``None`` when both orientations are
        permitted or when both are forbidden (caller decides what to do).
        """
        if (i, j) in self._required_dir_set:
            return (i, j)
        if (j, i) in self._required_dir_set:
            return (j, i)
        if self.layering is not None and i in self._layer_of and j in self._layer_of:
            li, lj = self._layer_of[i], self._layer_of[j]
            if li != lj:
                return (i, j) if li < lj else (j, i)
        ij_forbidden = (i, j) in self._forbidden_dir_set
        ji_forbidden = (j, i) in self._forbidden_dir_set
        if ij_forbidden ^ ji_forbidden:
            return (j, i) if ij_forbidden else (i, j)
        return None

    def filter_separating_set(self, i: str, j: str, candidates: list[str]) -> list[str]:
        """Drop conditioning-set candidates that lie in a layer strictly later than max(layer(i), layer(j))."""
        if self.layering is None:
            return list(candidates)
        layer_of = self._layer_of
        if i not in layer_of or j not in layer_of:
            return list(candidates)
        max_layer = max(layer_of[i], layer_of[j])
        return [c for c in candidates if c not in layer_of or layer_of[c] <= max_layer]

__post_init__()

Init dataclass.

Source code in mixpc/prior_knowledge.py

def __post_init__(self) -> None:
    """Init dataclass."""
    self._required_edges_set: set[frozenset[str]] = {frozenset(e) for e in self.required_edges}
    self._forbidden_edges_set: set[frozenset[str]] = {frozenset(e) for e in self.forbidden_edges}
    self._required_dir_set: set[Edge] = {(e[0], e[1]) for e in self.required_directions}
    self._forbidden_dir_set: set[Edge] = {(e[0], e[1]) for e in self.forbidden_directions}
    self._layer_of: dict[str, int] = {}
    if self.layering is not None:
        for idx, stage in enumerate(self.layering):
            for node in stage:
                self._layer_of[node] = idx

filter_separating_set(i, j, candidates)

Drop conditioning-set candidates that lie in a layer strictly later than max(layer(i), layer(j)).

Source code in mixpc/prior_knowledge.py

def filter_separating_set(self, i: str, j: str, candidates: list[str]) -> list[str]:
    """Drop conditioning-set candidates that lie in a layer strictly later than max(layer(i), layer(j))."""
    if self.layering is None:
        return list(candidates)
    layer_of = self._layer_of
    if i not in layer_of or j not in layer_of:
        return list(candidates)
    max_layer = max(layer_of[i], layer_of[j])
    return [c for c in candidates if c not in layer_of or layer_of[c] <= max_layer]

is_forbidden_direction(tail, head)

Whether the orientation tail -> head is forbidden by any hint or by layering.

Source code in mixpc/prior_knowledge.py

def is_forbidden_direction(self, tail: str, head: str) -> bool:
    """Whether the orientation ``tail -> head`` is forbidden by any hint or by layering."""
    if (tail, head) in self._forbidden_dir_set:
        return True
    if (head, tail) in self._required_dir_set:
        return True
    return (
        self.layering is not None
        and tail in self._layer_of
        and head in self._layer_of
        and self._layer_of[tail] > self._layer_of[head]
    )

is_forbidden_edge(i, j)

Whether the undirected edge {i, j} is blacklisted.

Source code in mixpc/prior_knowledge.py

def is_forbidden_edge(self, i: str, j: str) -> bool:
    """Whether the undirected edge ``{i, j}`` is blacklisted."""
    return frozenset((i, j)) in self._forbidden_edges_set

is_required_edge(i, j)

Whether the undirected edge {i, j} must appear (directly or via a required direction).

Source code in mixpc/prior_knowledge.py

def is_required_edge(self, i: str, j: str) -> bool:
    """Whether the undirected edge ``{i, j}`` must appear (directly or via a required direction)."""
    if frozenset((i, j)) in self._required_edges_set:
        return True
    return (i, j) in self._required_dir_set or (j, i) in self._required_dir_set

required_direction_for(i, j)

Return the uniquely allowed orientation of edge {i, j}, if any.

Resolution order: explicit required_direction → layering → forbidden_direction leaving exactly one valid side. Returns None when both orientations are permitted or when both are forbidden (caller decides what to do).

Source code in mixpc/prior_knowledge.py

def required_direction_for(self, i: str, j: str) -> Edge | None:
    """Return the uniquely allowed orientation of edge {i, j}, if any.

    Resolution order: explicit required_direction → layering → forbidden_direction
    leaving exactly one valid side. Returns ``None`` when both orientations are
    permitted or when both are forbidden (caller decides what to do).
    """
    if (i, j) in self._required_dir_set:
        return (i, j)
    if (j, i) in self._required_dir_set:
        return (j, i)
    if self.layering is not None and i in self._layer_of and j in self._layer_of:
        li, lj = self._layer_of[i], self._layer_of[j]
        if li != lj:
            return (i, j) if li < lj else (j, i)
    ij_forbidden = (i, j) in self._forbidden_dir_set
    ji_forbidden = (j, i) in self._forbidden_dir_set
    if ij_forbidden ^ ji_forbidden:
        return (j, i) if ij_forbidden else (i, j)
    return None

validate(nodes)

Check internal consistency and that every named node exists in nodes.

Source code in mixpc/prior_knowledge.py

def validate(self, nodes: set[str]) -> None:
    """Check internal consistency and that every named node exists in ``nodes``."""
    self._validate_node_membership(nodes)
    self._validate_edge_conflicts()
    if self.layering is not None:
        self._validate_layering()

Independence Tests

Bases: CItest

Nonparanormal Fisher Z conditional independence test for mixed continuous/ordinal data.

Uses :func:~mixpc.correlations.pairwise_latent_correlation to build a pairwise correlation matrix that automatically selects the right estimator for each variable pair:

• Both continuous → nonparanormal Spearman sin-transform.
• Both ordinal → polychoric MLE.
• Mixed → ad-hoc polyserial.

The partial correlation of X and Y given Z is then derived from the precision matrix of the joint correlation matrix, and Fisher's Z transform is applied.

Parameters:

Name	Type	Description	Default
n_levels_threshold	int	Variables with fewer unique values than this are treated as ordinal. Defaults to 20.	20
max_cor	float	Clip bound for individual pairwise correlations. Defaults to 0.9999.	0.9999

Source code in mixpc/independence_tests.py

class MixedFisherZ(CItest):
    """Nonparanormal Fisher Z conditional independence test for mixed continuous/ordinal data.

    Uses :func:`~mixpc.correlations.pairwise_latent_correlation` to build a pairwise
    correlation matrix that automatically selects the right estimator for each
    variable pair:

    - Both continuous → nonparanormal Spearman sin-transform.
    - Both ordinal → polychoric MLE.
    - Mixed → ad-hoc polyserial.

    The partial correlation of X and Y given Z is then derived from the
    precision matrix of the joint correlation matrix, and Fisher's Z transform
    is applied.

    Args:
        n_levels_threshold: Variables with fewer unique values than this are
            treated as ordinal. Defaults to 20.
        max_cor: Clip bound for individual pairwise correlations. Defaults to
            0.9999.
    """

    def __init__(self, n_levels_threshold: int = 20, max_cor: float = 0.9999) -> None:
        """Init. Variables with < n_levels_threshold unique values are treated as ordinal."""
        self._n_levels_threshold = n_levels_threshold
        self._max_cor = max_cor

    @staticmethod
    def _as_1d(data: np.ndarray | pd.DataFrame | pd.Series) -> np.ndarray:
        arr = data.to_numpy() if isinstance(data, pd.DataFrame | pd.Series) else np.asarray(data)
        return arr.ravel()

    def _build_corr_matrix(self, cols: list[np.ndarray]) -> np.ndarray:
        """Build a full pairwise correlation matrix from a list of 1-D arrays."""
        k = len(cols)
        corr = np.eye(k)
        for i in range(k):
            for j in range(i + 1, k):
                r = pairwise_latent_correlation(
                    cols[i],
                    cols[j],
                    max_cor=self._max_cor,
                    n_levels_threshold=self._n_levels_threshold,
                )
                corr[i, j] = corr[j, i] = r
        return corr

    def test(
        self,
        x_data: np.ndarray | pd.DataFrame | pd.Series,
        y_data: np.ndarray | pd.DataFrame | pd.Series,
        z_data: np.ndarray | pd.DataFrame | pd.Series | None = None,
        corr_threshold: float = 0.999,
    ) -> tuple[float, float]:
        """Test conditional independence of X and Y given Z.

        Args:
            x_data: Variable X — shape (n,) or (n, 1).
            y_data: Variable Y — shape (n,) or (n, 1).
            z_data: Conditioning set — shape (n, k) or None for marginal test.
            corr_threshold: Clip bound applied to the partial correlation
                before the Fisher Z transform.

        Returns:
            (test_statistic, p_value)
        """
        self._check_input(x_data, y_data, z_data)

        x_arr = self._as_1d(x_data)
        y_arr = self._as_1d(y_data)
        n = x_arr.shape[0]

        if z_data is None:
            # Marginal test: direct pairwise correlation
            r = pairwise_latent_correlation(
                x_arr, y_arr,
                max_cor=corr_threshold,
                n_levels_threshold=self._n_levels_threshold,
            )
            sep_set_length = 0
            cols = [x_arr, y_arr]
        else:
            z_arr = z_data.to_numpy() if isinstance(z_data, pd.DataFrame | pd.Series) else np.asarray(z_data)
            if z_arr.ndim == 1:
                z_arr = z_arr[:, np.newaxis]
            sep_set_length = z_arr.shape[1]

            # Build correlation matrix for [X, Y, Z_1, ..., Z_k]
            cols = [x_arr, y_arr] + [z_arr[:, i] for i in range(sep_set_length)]
            corr_mat = self._build_corr_matrix(cols)
            corr_mat = _make_positive_definite(corr_mat)

            try:
                precision = np.linalg.inv(corr_mat)
            except np.linalg.LinAlgError as exc:
                raise ValueError("Correlation matrix is singular; check for collinearities.") from exc

            # Partial correlation r(X,Y|Z) via precision matrix
            r = -precision[0, 1] / np.sqrt(np.abs(precision[0, 0] * precision[1, 1]))

        r = float(np.clip(r, -corr_threshold, corr_threshold))
        factor = np.sqrt(n - sep_set_length - 3)
        z_stat = factor * 0.5 * np.log((1 + r) / (1 - r))
        p_value = float(2 * (1 - norm.cdf(abs(z_stat))))

        return (float(z_stat), p_value)

__init__(n_levels_threshold=20, max_cor=0.9999)

Init. Variables with < n_levels_threshold unique values are treated as ordinal.

Source code in mixpc/independence_tests.py

def __init__(self, n_levels_threshold: int = 20, max_cor: float = 0.9999) -> None:
    """Init. Variables with < n_levels_threshold unique values are treated as ordinal."""
    self._n_levels_threshold = n_levels_threshold
    self._max_cor = max_cor

test(x_data, y_data, z_data=None, corr_threshold=0.999)

Test conditional independence of X and Y given Z.

Parameters:

Name	Type	Description	Default
x_data	ndarray | DataFrame | Series	Variable X — shape (n,) or (n, 1).	required
y_data	ndarray | DataFrame | Series	Variable Y — shape (n,) or (n, 1).	required
z_data	ndarray | DataFrame | Series | None	Conditioning set — shape (n, k) or None for marginal test.	None
corr_threshold	float	Clip bound applied to the partial correlation before the Fisher Z transform.	0.999

Returns:

Type	Description
tuple[float, float]	(test_statistic, p_value)

Source code in mixpc/independence_tests.py

def test(
    self,
    x_data: np.ndarray | pd.DataFrame | pd.Series,
    y_data: np.ndarray | pd.DataFrame | pd.Series,
    z_data: np.ndarray | pd.DataFrame | pd.Series | None = None,
    corr_threshold: float = 0.999,
) -> tuple[float, float]:
    """Test conditional independence of X and Y given Z.

    Args:
        x_data: Variable X — shape (n,) or (n, 1).
        y_data: Variable Y — shape (n,) or (n, 1).
        z_data: Conditioning set — shape (n, k) or None for marginal test.
        corr_threshold: Clip bound applied to the partial correlation
            before the Fisher Z transform.

    Returns:
        (test_statistic, p_value)
    """
    self._check_input(x_data, y_data, z_data)

    x_arr = self._as_1d(x_data)
    y_arr = self._as_1d(y_data)
    n = x_arr.shape[0]

    if z_data is None:
        # Marginal test: direct pairwise correlation
        r = pairwise_latent_correlation(
            x_arr, y_arr,
            max_cor=corr_threshold,
            n_levels_threshold=self._n_levels_threshold,
        )
        sep_set_length = 0
        cols = [x_arr, y_arr]
    else:
        z_arr = z_data.to_numpy() if isinstance(z_data, pd.DataFrame | pd.Series) else np.asarray(z_data)
        if z_arr.ndim == 1:
            z_arr = z_arr[:, np.newaxis]
        sep_set_length = z_arr.shape[1]

        # Build correlation matrix for [X, Y, Z_1, ..., Z_k]
        cols = [x_arr, y_arr] + [z_arr[:, i] for i in range(sep_set_length)]
        corr_mat = self._build_corr_matrix(cols)
        corr_mat = _make_positive_definite(corr_mat)

        try:
            precision = np.linalg.inv(corr_mat)
        except np.linalg.LinAlgError as exc:
            raise ValueError("Correlation matrix is singular; check for collinearities.") from exc

        # Partial correlation r(X,Y|Z) via precision matrix
        r = -precision[0, 1] / np.sqrt(np.abs(precision[0, 0] * precision[1, 1]))

    r = float(np.clip(r, -corr_threshold, corr_threshold))
    factor = np.sqrt(n - sep_set_length - 3)
    z_stat = factor * 0.5 * np.log((1 + r) / (1 - r))
    p_value = float(2 * (1 - norm.cdf(abs(z_stat))))

    return (float(z_stat), p_value)

Correlation Measures

Bases: CorrelationMeasure

MLE polychoric correlation between two ordinal variables.

Parameters:

Name	Type	Description	Default
max_cor	float	Clip bound for the estimate. Defaults to 0.9999.	0.9999
solver	Literal['newton', 'brent']	"brent" (default, faster for ≥4 categories) or "newton" (Fisher scoring, faster for binary/ternary).	'brent'
max_iter	int	Max iterations for Newton solver.	100
tol	float	Convergence tolerance for Newton solver.	1e-10

Source code in mixpc/correlations.py

class PolychoricCorrelation(CorrelationMeasure):
    """MLE polychoric correlation between two ordinal variables.

    Args:
        max_cor: Clip bound for the estimate. Defaults to 0.9999.
        solver: ``"brent"`` (default, faster for ≥4 categories) or
            ``"newton"`` (Fisher scoring, faster for binary/ternary).
        max_iter: Max iterations for Newton solver.
        tol: Convergence tolerance for Newton solver.
    """

    def __init__(
        self,
        max_cor: float = 0.9999,
        solver: Literal["newton", "brent"] = "brent",
        max_iter: int = 100,
        tol: float = 1e-10,
    ) -> None:
        """Init. solver: 'brent' (default) or 'newton' (Fisher scoring)."""
        super().__init__(max_cor=max_cor)
        if solver not in {"newton", "brent"}:
            raise ValueError(f"`solver` must be 'newton' or 'brent', got '{solver}'.")
        self._solver = solver
        self._max_iter = max_iter
        self._tol = tol

    def fit(self, x: np.ndarray, y: np.ndarray) -> PolychoricCorrelation:
        """Fit polychoric correlation to two ordinal arrays. Returns self."""
        x_arr, y_arr = self._prepare(x, y)
        _validate_ordinal(x_arr, "x")
        _validate_ordinal(y_arr, "y")
        self._correlation = self._clip(self._polychoric(x_arr, y_arr))
        return self

    def _polychoric(self, x: np.ndarray, y: np.ndarray) -> float:
        n = x.size
        ux, uy = np.unique(x), np.unique(y)
        n_rs = np.array([[np.sum((x == xi) & (y == yj)) for yj in uy] for xi in ux], dtype=float)
        assert n_rs.sum() == n
        tx, ty = _thresholds(x), _thresholds(y)
        if self._solver == "newton":
            return self._polychoric_newton(n_rs, tx, ty, ux, uy)
        return self._polychoric_brent(n_rs, tx, ty, ux, uy)

    def _polychoric_newton(
        self, n_rs: np.ndarray, tx: np.ndarray, ty: np.ndarray, ux: np.ndarray, uy: np.ndarray
    ) -> float:
        rho = 0.0
        bound = self._max_cor
        score_val = 0.0
        for iteration in range(self._max_iter):
            score_val = info_val = 0.0
            for i in range(len(ux)):
                for j in range(len(uy)):
                    lower = (float(tx[i]), float(ty[j]))
                    upper = (float(tx[i + 1]), float(ty[j + 1]))
                    p = max(_pi_rs(lower=lower, upper=upper, corr=rho), _CELL_FLOOR)
                    dp = _pi_rs_derivative(lower=np.array(lower), upper=np.array(upper), corr=rho)
                    ratio = dp / p
                    score_val += n_rs[i, j] * ratio
                    info_val += n_rs[i, j] * ratio**2
            if abs(score_val) < self._tol:
                break
            if info_val < 1e-14:
                logger.warning(
                    "Fisher information ≈ 0 at ρ=%.4f after %d iter; returning current estimate.", rho, iteration
                )
                break
            step = score_val / info_val
            rho_new = float(np.clip(rho + step, -bound, bound))
            if abs(rho_new - rho) < self._tol:
                rho = rho_new
                break
            rho = rho_new
        else:
            logger.warning(
                "Fisher scoring did not converge in %d iterations (|score|=%.2e).", self._max_iter, abs(score_val)
            )
        return rho

    def _polychoric_brent(
        self, n_rs: np.ndarray, tx: np.ndarray, ty: np.ndarray, ux: np.ndarray, uy: np.ndarray
    ) -> float:
        bound = self._max_cor

        def neg_log_likelihood(rho: float) -> float:
            total = 0.0
            for i in range(len(ux)):
                for j in range(len(uy)):
                    if n_rs[i, j] == 0:
                        continue
                    lower = (float(tx[i]), float(ty[j]))
                    upper = (float(tx[i + 1]), float(ty[j + 1]))
                    p = max(_pi_rs(lower=lower, upper=upper, corr=rho), _CELL_FLOOR)
                    total += n_rs[i, j] * np.log(p)
            return -total

        result = minimize_scalar(neg_log_likelihood, bounds=(-bound, bound), method="bounded")
        return float(result.x)

__init__(max_cor=0.9999, solver='brent', max_iter=100, tol=1e-10)

Init. solver: 'brent' (default) or 'newton' (Fisher scoring).

Source code in mixpc/correlations.py

def __init__(
    self,
    max_cor: float = 0.9999,
    solver: Literal["newton", "brent"] = "brent",
    max_iter: int = 100,
    tol: float = 1e-10,
) -> None:
    """Init. solver: 'brent' (default) or 'newton' (Fisher scoring)."""
    super().__init__(max_cor=max_cor)
    if solver not in {"newton", "brent"}:
        raise ValueError(f"`solver` must be 'newton' or 'brent', got '{solver}'.")
    self._solver = solver
    self._max_iter = max_iter
    self._tol = tol

fit(x, y)

Fit polychoric correlation to two ordinal arrays. Returns self.

Source code in mixpc/correlations.py

def fit(self, x: np.ndarray, y: np.ndarray) -> PolychoricCorrelation:
    """Fit polychoric correlation to two ordinal arrays. Returns self."""
    x_arr, y_arr = self._prepare(x, y)
    _validate_ordinal(x_arr, "x")
    _validate_ordinal(y_arr, "y")
    self._correlation = self._clip(self._polychoric(x_arr, y_arr))
    return self

Bases: CorrelationMeasure

Ad-hoc polyserial correlation between one continuous and one ordinal variable.

Parameters:

Name	Type	Description	Default
max_cor	float	Clip bound. Defaults to 0.9999.	0.9999
n_levels_threshold	int	Variables with fewer unique values are treated as ordinal. Defaults to 20.	20

Source code in mixpc/correlations.py

class PolyserialCorrelation(CorrelationMeasure):
    """Ad-hoc polyserial correlation between one continuous and one ordinal variable.

    Args:
        max_cor: Clip bound. Defaults to 0.9999.
        n_levels_threshold: Variables with fewer unique values are treated as
            ordinal. Defaults to 20.
    """

    def __init__(self, max_cor: float = 0.9999, n_levels_threshold: int = 20) -> None:
        """Init. Variables with < n_levels_threshold unique values are treated as ordinal."""
        super().__init__(max_cor=max_cor)
        if n_levels_threshold < 2:
            raise ValueError("`n_levels_threshold` must be ≥ 2.")
        self._n_levels_threshold = n_levels_threshold

    def fit(self, x: np.ndarray, y: np.ndarray) -> PolyserialCorrelation:
        """Fit polyserial correlation to a continuous/ordinal pair. Returns self."""
        x_arr, y_arr = self._prepare(x, y)
        x_is_ord = len(np.unique(x_arr)) < self._n_levels_threshold
        y_is_ord = len(np.unique(y_arr)) < self._n_levels_threshold
        if not x_is_ord and not y_is_ord:
            raise ValueError("Both variables appear continuous; use a Spearman/NPN estimator instead.")
        if x_is_ord and y_is_ord:
            raise ValueError("Both variables appear ordinal; use PolychoricCorrelation instead.")
        cont_arr, ord_arr = (y_arr, x_arr) if x_is_ord else (x_arr, y_arr)
        cont_name, ord_name = ("y", "x") if x_is_ord else ("x", "y")
        _validate_continuous(cont_arr, cont_name)
        _validate_ordinal(ord_arr, ord_name)
        self._correlation = self._clip(self._polyserial(cont_arr, ord_arr))
        return self

    def _polyserial(self, cont: np.ndarray, disc: np.ndarray) -> float:
        unique_vals = np.sort(np.unique(disc).astype(float))
        threshold_estimate = _thresholds(disc)
        interior_thresholds = threshold_estimate[1:-1]
        value_diff = np.diff(unique_vals)
        lambda_val = float(np.sum(stats.norm.pdf(interior_thresholds) * value_diff))
        if lambda_val == 0:
            raise ValueError("Denominator λ in polyserial estimator is zero.")
        s_disc = float(np.std(disc.astype(float), ddof=1))
        r = _npn_pearson(cont, disc)
        return r * s_disc / lambda_val

__init__(max_cor=0.9999, n_levels_threshold=20)

Init. Variables with < n_levels_threshold unique values are treated as ordinal.

Source code in mixpc/correlations.py

def __init__(self, max_cor: float = 0.9999, n_levels_threshold: int = 20) -> None:
    """Init. Variables with < n_levels_threshold unique values are treated as ordinal."""
    super().__init__(max_cor=max_cor)
    if n_levels_threshold < 2:
        raise ValueError("`n_levels_threshold` must be ≥ 2.")
    self._n_levels_threshold = n_levels_threshold

fit(x, y)

Fit polyserial correlation to a continuous/ordinal pair. Returns self.

Source code in mixpc/correlations.py

def fit(self, x: np.ndarray, y: np.ndarray) -> PolyserialCorrelation:
    """Fit polyserial correlation to a continuous/ordinal pair. Returns self."""
    x_arr, y_arr = self._prepare(x, y)
    x_is_ord = len(np.unique(x_arr)) < self._n_levels_threshold
    y_is_ord = len(np.unique(y_arr)) < self._n_levels_threshold
    if not x_is_ord and not y_is_ord:
        raise ValueError("Both variables appear continuous; use a Spearman/NPN estimator instead.")
    if x_is_ord and y_is_ord:
        raise ValueError("Both variables appear ordinal; use PolychoricCorrelation instead.")
    cont_arr, ord_arr = (y_arr, x_arr) if x_is_ord else (x_arr, y_arr)
    cont_name, ord_name = ("y", "x") if x_is_ord else ("x", "y")
    _validate_continuous(cont_arr, cont_name)
    _validate_ordinal(ord_arr, ord_name)
    self._correlation = self._clip(self._polyserial(cont_arr, ord_arr))
    return self

Dispatch to the appropriate correlation estimator based on variable types.

• Both continuous → nonparanormal Spearman sin-transform.
• Both ordinal → polychoric MLE.
• Mixed → ad-hoc polyserial.

Parameters:

Name	Type	Description	Default
x	ndarray	First variable.	required
y	ndarray	Second variable (same length as x).	required
max_cor	float	Clip bound for the result.	0.9999
n_levels_threshold	int	Unique-value count below which a variable is treated as ordinal.	20
verbose	bool	Log which estimator was selected.	False

Returns:

Type	Description
float	Correlation estimate in [−1, 1].

Source code in mixpc/correlations.py

def pairwise_latent_correlation(
    x: np.ndarray,
    y: np.ndarray,
    *,
    max_cor: float = 0.9999,
    n_levels_threshold: int = 20,
    verbose: bool = False,
) -> float:
    """Dispatch to the appropriate correlation estimator based on variable types.

    - Both continuous → nonparanormal Spearman sin-transform.
    - Both ordinal → polychoric MLE.
    - Mixed → ad-hoc polyserial.

    Args:
        x: First variable.
        y: Second variable (same length as x).
        max_cor: Clip bound for the result.
        n_levels_threshold: Unique-value count below which a variable is
            treated as ordinal.
        verbose: Log which estimator was selected.

    Returns:
        Correlation estimate in [−1, 1].
    """
    x_arr, y_arr = _to_array(x), _to_array(y)
    _validate_pair(x_arr, y_arr)

    x_is_disc = len(np.unique(x_arr)) < n_levels_threshold
    y_is_disc = len(np.unique(y_arr)) < n_levels_threshold

    if not x_is_disc and not y_is_disc:
        if verbose:
            logger.info("Both continuous — using nonparanormal Spearman.")
        rho = spearman(x_arr, y_arr)
        return float(np.clip(2 * np.sin(np.pi / 6 * rho), -max_cor, max_cor))

    if x_is_disc and y_is_disc:
        if verbose:
            logger.info("Both ordinal — using polychoric correlation.")
        return PolychoricCorrelation(max_cor=max_cor).fit(x_arr, y_arr).correlation

    if verbose:
        logger.info("Mixed pair — using polyserial correlation.")
    return PolyserialCorrelation(max_cor=max_cor, n_levels_threshold=n_levels_threshold).fit(x_arr, y_arr).correlation

Winsorized nonparanormal transformation (Liu et al. 2009).

Parameters:

Name	Type	Description	Default
x	ndarray	1-D numeric array (≥ 2 observations).	required

Returns:

Type	Description
ndarray	Transformed array scaled to unit variance.

Source code in mixpc/correlations.py

def f_hat(x: np.ndarray) -> np.ndarray:
    """Winsorized nonparanormal transformation (Liu et al. 2009).

    Args:
        x: 1-D numeric array (≥ 2 observations).

    Returns:
        Transformed array scaled to unit variance.
    """
    x_arr = _to_array(x)
    if x_arr.size < 2:
        raise ValueError("f_hat requires at least 2 observations.")
    return _f_hat(x_arr)

Graph Classes

Bases: GRAPH

Class for dealing with partially directed graph i.e.

graphs that contain both directed and undirected edges.

Source code in mixpc/graphs.py

class PDAG(GRAPH):
    """Class for dealing with partially directed graph i.e.

    graphs that contain both directed and undirected edges.
    """

    def __init__(
        self,
        nodes: list[str] | None = None,
        dir_edges: list[tuple[str, str]] | None = None,
        undir_edges: list[tuple[str, str]] | None = None,
    ) -> None:
        """PDAG constructor.

        Args:
            nodes (list[str] | None, optional): Nodes in the PDAG. Defaults to None.
            dir_edges (list[tuple[str,str]] | None, optional): directed edges. Defaults to None.
            undir_edges (list[tuple[str,str]] | None, optional): undirected edges. Defaults to None.
        """
        if nodes is None:
            nodes = []
        if dir_edges is None:
            dir_edges = []
        if undir_edges is None:
            undir_edges = []

        self._nodes = set(nodes)
        self._undir_edges: set[tuple[str, str]] = set()
        self._dir_edges: set[tuple[str, str]] = set()
        self._parents: defaultdict[str, set[str]] = defaultdict(set)
        self._children: defaultdict[str, set[str]] = defaultdict(set)
        self._neighbors: defaultdict[str, set[str]] = defaultdict(set)
        self._undirected_neighbors: defaultdict[str, set[str]] = defaultdict(set)

        for dir_edge in dir_edges:
            self._add_dir_edge(*dir_edge)
        for unir_edge in undir_edges:
            self._add_undir_edge(*unir_edge)

    def _add_dir_edge(self, i: str, j: str) -> None:
        self._nodes.add(i)
        self._nodes.add(j)
        self._dir_edges.add((i, j))

        self._neighbors[i].add(j)
        self._neighbors[j].add(i)

        self._children[i].add(j)
        self._parents[j].add(i)

    def _add_undir_edge(self, i: str, j: str) -> None:
        self._nodes.add(i)
        self._nodes.add(j)
        self._undir_edges.add((i, j))

        self._neighbors[i].add(j)
        self._neighbors[j].add(i)

        self._undirected_neighbors[i].add(j)
        self._undirected_neighbors[j].add(i)

    def children(self, node: str) -> set[str]:
        """Gives all children of node `node`.

        Args:
            node (str): node in current PDAG.

        Returns:
            set: set of children.
        """
        if node in self._children:
            return self._children[node]
        else:
            return set()

    def parents(self, node: str) -> set[str]:
        """Gives all parents of node `node`.

        Args:
            node (str): node in current PDAG.

        Returns:
            set: set of parents.
        """
        if node in self._parents:
            return self._parents[node]
        else:
            return set()

    def neighbors(self, node: str) -> set[str]:
        """Gives all neighbors of node `node`.

        Args:
            node (str): node in current PDAG.

        Returns:
            set: set of neighbors.
        """
        if node in self._neighbors:
            return self._neighbors[node]
        else:
            return set()

    def undir_neighbors(self, node: str) -> set[str]:
        """Gives all undirected neighbors of node `node`.

        Args:
            node (str): node in current PDAG.

        Returns:
            set: set of undirected neighbors.
        """
        if node in self._undirected_neighbors:
            return self._undirected_neighbors[node]
        else:
            return set()

    def is_adjacent(self, i: str, j: str) -> bool:
        """Return True if the graph contains an directed or undirected edge between i and j.

        Args:
            i (str): node i.
            j (str): node j.

        Returns:
            bool: True if i-j or i->j or i<-j
        """
        return (
            (i, j) in self._dir_edges
            or (j, i) in self._dir_edges
            or (i, j) in self._undir_edges
            or (j, i) in self._undir_edges
        )

    def is_clique(self, potential_clique: set[str]) -> bool:
        """Check every pair of node X potential_clique is adjacent."""
        return all(self.is_adjacent(i, j) for i, j in combinations(potential_clique, 2))

    @classmethod
    def from_pandas_adjacency(cls, pd_amat: pd.DataFrame) -> PDAG:
        """Build PDAG from a Pandas adjacency matrix.

        Args:
            pd_amat (pd.DataFrame): input adjacency matrix.

        Returns:
            PDAG
        """
        assert pd_amat.shape[0] == pd_amat.shape[1]
        nodes = list(pd_amat.columns)

        all_connections = []
        start, end = np.where(pd_amat != 0)
        for idx, _ in enumerate(start):
            all_connections.append((pd_amat.columns[start[idx]], pd_amat.columns[end[idx]]))

        temp = [set(i) for i in all_connections]
        temp2 = [arc for arc in all_connections if temp.count(set(arc)) > 1]
        undir_edges = [tuple(item) for item in set(frozenset(item) for item in temp2)]

        dir_edges = [edge for edge in all_connections if edge not in temp2]

        return PDAG(nodes=nodes, dir_edges=dir_edges, undir_edges=undir_edges)

    def remove_edge(self, i: str, j: str) -> None:
        """Removes edge in question.

        Args:
            i (str): tail
            j (str): head

        Raises:
            AssertionError: if edge does not exist
        """
        if (i, j) not in self._dir_edges and (i, j) not in self._undir_edges:
            raise AssertionError("Edge does not exist in current PDAG")

        self._undir_edges.discard((i, j))
        self._dir_edges.discard((i, j))
        self._children[i].discard(j)
        self._parents[j].discard(i)
        self._neighbors[i].discard(j)
        self._neighbors[j].discard(i)
        self._undirected_neighbors[i].discard(j)
        self._undirected_neighbors[j].discard(i)

    def undir_to_dir_edge(self, tail: str, head: str) -> None:
        """Takes a undirected edge and turns it into a directed one.

        tail indicates the starting node of the edge and head the end node, i.e.
        tail -> head.

        Args:
            tail (str): starting node
            head (str): end node

        Raises:
            AssertionError: if edge does not exist or is not undirected.
        """
        if (tail, head) not in self._undir_edges and (head, tail) not in self._undir_edges:
            raise AssertionError("Edge seems not to be undirected or even there at all.")
        self._undir_edges.discard((tail, head))
        self._undir_edges.discard((head, tail))
        self._neighbors[tail].discard(head)
        self._neighbors[head].discard(tail)
        self._undirected_neighbors[tail].discard(head)
        self._undirected_neighbors[head].discard(tail)

        self._add_dir_edge(i=tail, j=head)

    def remove_node(self, node: str) -> None:
        """Remove a node from the graph.

        Args:
            node (str): node to remove
        """
        self._nodes.remove(node)

        self._dir_edges = {(i, j) for i, j in self._dir_edges if node not in {i, j}}

        self._undir_edges = {(i, j) for i, j in self._undir_edges if node not in {i, j}}

        for child in self._children[node]:
            self._parents[child].remove(node)
            self._neighbors[child].remove(node)

        for parent in self._parents[node]:
            self._children[parent].remove(node)
            self._neighbors[parent].remove(node)

        for u_nbr in self._undirected_neighbors[node]:
            self._undirected_neighbors[u_nbr].remove(node)
            self._neighbors[u_nbr].remove(node)

        self._parents.pop(node, "I was never here")
        self._children.pop(node, "I was never here")
        self._neighbors.pop(node, "I was never here")
        self._undirected_neighbors.pop(node, "I was never here")

    def to_dag(self) -> nx.DiGraph:
        r"""Algorithm as described in Chickering (2002).

            1. From PDAG P create DAG G containing all directed edges from P
            2. Repeat the following: Select node v in P s.t.
                i. v has no outgoing edges (children) i.e. \\(ch(v) = \\emptyset \\)

                ii. \\(neigh(v) \\neq \\emptyset\\)
                    Then \\( (pa(v) \\cup (neigh(v) \\) form a clique.
                    For each v that is in a clique and is part of an undirected edge in P
                    i.e. w - v, insert a directed edge w -> v in G.
                    Remove v and all incident edges from P and continue with next node.
                    Until all nodes have been deleted from P.

        Returns:
            nx.DiGraph: DAG that belongs to the MEC implied by the PDAG
        """
        pdag = self.copy()

        dag = nx.DiGraph()
        dag.add_nodes_from(pdag.nodes)
        dag.add_edges_from(pdag.dir_edges)

        if pdag.num_undir_edges == 0:
            return dag
        else:
            while pdag.num_nodes > 0:
                # find node with (1) no directed outgoing edges and
                #                (2) the set of undirected neighbors is either empty or
                #                    undirected neighbors + parents of X are a clique
                found = False
                for node in pdag.nodes:
                    children = pdag.children(node)
                    neighbors = pdag.neighbors(node)
                    # pdag._undirected_neighbors[node]
                    parents = pdag.parents(node)
                    potential_clique_members = neighbors.union(parents)

                    is_clique = pdag.is_clique(potential_clique_members)

                    if not children and (not neighbors or is_clique):
                        found = True
                        # add all edges of node as outgoing edges to dag
                        for edge in pdag.undir_edges:
                            if node in edge:
                                incident_node = next(iter(set(edge) - {node}))
                                dag.add_edge(incident_node, node)

                        pdag.remove_node(node)
                        break

                if not found:
                    logger.warning("PDAG not extendible: Random DAG on skeleton drawn.")

                    dag = nx.from_pandas_adjacency(self._amat_to_dag(), create_using=nx.DiGraph)

                    break

            return dag

    @property
    def adjacency_matrix(self) -> pd.DataFrame:
        """Returns adjacency matrix.

        The i,jth entry being one indicates that there is an edge
        from i to j. A zero indicates that there is no edge.

        Returns:
            pd.DataFrame: adjacency matrix
        """
        amat = pd.DataFrame(
            np.zeros([self.num_nodes, self.num_nodes]),
            index=self.nodes,
            columns=self.nodes,
        )
        for edge in self.dir_edges:
            amat.loc[edge] = 1
        for edge in self.undir_edges:
            amat.loc[edge] = amat.loc[edge[::-1]] = 1
        return amat

    @property
    def causal_order(self) -> None:
        """Causal order is None.

        This is because PDAGs only allow for a partial causal order.

        Returns:
            None: None
        """
        return None

    def _amat_to_dag(self) -> pd.DataFrame:
        """Transform the adjacency matrix of an PDAG to the adjacency matrix.

            of SOME DAG in the Markov equivalence class.

        Returns:
            pd.DataFrame: DAG, a member of the MEC.
        """
        pdag_amat = self.adjacency_matrix.to_numpy()

        p = pdag_amat.shape[0]
        # amat to skel
        skel = pdag_amat + pdag_amat.T
        skel[np.where(skel > 1)] = 1
        # permute skel
        rng = np.random.default_rng()
        permute_ord = rng.choice(a=p, size=p, replace=False)
        skel = skel[:, permute_ord][permute_ord]

        # skel to dag
        for i in range(1, p):
            for j in range(0, i + 1):
                if skel[i, j] == 1:
                    skel[i, j] = 0

        # inverse permutation
        i_ord = np.sort(permute_ord)
        skel = skel[:, i_ord][i_ord]
        return pd.DataFrame(
            skel,
            index=self.adjacency_matrix.index,
            columns=self.adjacency_matrix.columns,
        )

    def vstructs(self) -> set[tuple[str, str]]:
        """Retrieve v-structures.

        Returns:
            set: set of all v-structures
        """
        vstructures = set()
        for node in self._nodes:
            for p1, p2 in combinations(self._parents[node], 2):
                if p1 not in self._parents[p2] and p2 not in self._parents[p1]:
                    vstructures.add((p1, node))
                    vstructures.add((p2, node))
        return vstructures

    def copy(self) -> PDAG:
        """Return a copy of the graph."""
        return PDAG(
            nodes=list(self._nodes),
            dir_edges=list(self._dir_edges),
            undir_edges=list(self._undir_edges),
        )

    def show(self) -> None:
        """Plot PDAG."""
        graph = self.to_networkx()
        pos = nx.circular_layout(graph)
        nx.draw(graph, pos=pos, with_labels=True)

    def to_networkx(self) -> nx.MultiDiGraph:
        """Convert to networkx graph.

        Returns:
            nx.MultiDiGraph: Graph with directed and undirected edges.
        """
        nx_pdag = nx.MultiDiGraph()
        nx_pdag.add_nodes_from(self.nodes)
        nx_pdag.add_edges_from(self.dir_edges)
        for edge in self.undir_edges:
            nx_pdag.add_edge(*edge)
            nx_pdag.add_edge(*edge[::-1])

        return nx_pdag

    def _meek_mec_enumeration(self, pdag: PDAG, dag_list: list[DAG]) -> None:
        """Apply Meek's MEC enumeration algorithm.

        Args:
            pdag (PDAG): partially directed graph in question.
            dag_list (list): list of currently found DAGs.

        References:
            Wienöbst, Marcel, et al. "Efficient enumeration of Markov equivalent DAGs."
            Proceedings of the AAAI Conference on Artificial Intelligence.
            Vol. 37. No. 10. 2023.
        """
        g_copy = pdag.copy()
        g_copy = self._apply_meek_rules(g_copy)  # Apply Meek rules

        if not g_copy.undir_edges:
            # makes sure that flaoting nodes are preserved
            new_member = DAG()
            new_member.add_nodes_from(g_copy.nodes)
            new_member.add_edges_from(g_copy.dir_edges)
            dag_list.append(new_member)
            return  # Add DAG to current list

        i, j = g_copy.undir_edges[0]  # Take first undirected edge

        # Recursion first orientation:
        g_copy.undir_to_dir_edge(i, j)
        self._meek_mec_enumeration(pdag=g_copy, dag_list=dag_list)
        g_copy.remove_edge(i, j)

        # Recursion second orientation
        g_copy._add_dir_edge(j, i)
        self._meek_mec_enumeration(pdag=g_copy, dag_list=dag_list)

    def to_allDAGs(self) -> list[DAG]:  # noqa: N802
        """Recursion algorithm which recursively applies the following steps.

            1. Orient the first undirected edge found.
            2. Apply Meek rules.
            3. Recurse with each direction of the oriented edge.
        This corresponds to Algorithm 2 in Wienöbst et al. (2023).

        References:
            Wienöbst, Marcel, et al. "Efficient enumeration of Markov equivalent DAGs."
            Proceedings of the AAAI Conference on Artificial Intelligence.
            Vol. 37. No. 10. 2023.
        """
        all_dags: list[DAG] = []
        self._meek_mec_enumeration(pdag=self, dag_list=all_dags)
        return all_dags

    # use Meek's cpdag2alldag
    def _apply_meek_rules(self, pdag: PDAG) -> PDAG:
        """Apply all four Meek rules to a PDAG turning it into a CPDAG.

        Args:
            pdag (PDAG): PDAG to complete

        Returns:
            PDAG: completed PDAG.
        """
        # Apply Meek Rules
        cpdag = pdag.copy()
        cpdag = rule_1(pdag=cpdag)
        cpdag = rule_2(pdag=cpdag)
        cpdag = rule_3(pdag=cpdag)
        cpdag = rule_4(pdag=cpdag)
        return cpdag

    def to_random_dag(self) -> DAG:
        """Provides a random DAG residing in the MEC.

        Returns:
            nx.DiGraph: random DAG living in MEC
        """
        to_dag_candidate = self.copy()
        rng = np.random.default_rng()

        while to_dag_candidate.num_undir_edges > 0:
            chosen_edge = to_dag_candidate.undir_edges[rng.choice(to_dag_candidate.num_undir_edges)]
            choose_orientation = [chosen_edge, chosen_edge[::-1]]
            node_i, node_j = choose_orientation[rng.choice(len(choose_orientation))]

            to_dag_candidate.undir_to_dir_edge(tail=node_i, head=node_j)
            to_dag_candidate = to_dag_candidate._apply_meek_rules(pdag=to_dag_candidate)

        return DAG.from_pandas_adjacency(to_dag_candidate.adjacency_matrix)

    @property
    def nodes(self) -> list[str]:
        """Get all nods in current PDAG.

        Returns:
            list: list of nodes.
        """
        return sorted(list(self._nodes))

    @property
    def num_nodes(self) -> int:
        """Number of nodes in current PDAG.

        Returns:
            int: Number of nodes
        """
        return len(self._nodes)

    @property
    def num_undir_edges(self) -> int:
        """Number of undirected edges in current PDAG.

        Returns:
            int: Number of undirected edges
        """
        return len(self._undir_edges)

    @property
    def num_dir_edges(self) -> int:
        """Number of directed edges in current PDAG.

        Returns:
            int: Number of directed edges
        """
        return len(self._dir_edges)

    @property
    def num_adjacencies(self) -> int:
        """Number of adjacent nodes in current PDAG.

        Returns:
            int: Number of adjacent nodes
        """
        return self.num_undir_edges + self.num_dir_edges

    @property
    def undir_edges(self) -> list[tuple[str, str]]:
        """Gives all undirected edges in current PDAG.

        Returns:
            list[tuple[str,str]]: List of undirected edges, sorted for determinism.
        """
        return sorted(self._undir_edges)

    @property
    def dir_edges(self) -> list[tuple[str, str]]:
        """Gives all directed edges in current PDAG.

        Returns:
            list[tuple[str,str]]: List of directed edges, sorted for determinism.
        """
        return sorted(self._dir_edges)

adjacency_matrix property

Returns adjacency matrix.

The i,jth entry being one indicates that there is an edge from i to j. A zero indicates that there is no edge.

Returns:

Type	Description
DataFrame	pd.DataFrame: adjacency matrix

causal_order property

Causal order is None.

This is because PDAGs only allow for a partial causal order.

Returns:

Name	Type	Description
None	None	None

dir_edges property

Gives all directed edges in current PDAG.

Returns:

Type	Description
list[tuple[str, str]]	list[tuple[str,str]]: List of directed edges, sorted for determinism.

nodes property

Get all nods in current PDAG.

Returns:

Name	Type	Description
list	list[str]	list of nodes.

num_adjacencies property

Number of adjacent nodes in current PDAG.

Returns:

Name	Type	Description
int	int	Number of adjacent nodes

num_dir_edges property

Number of directed edges in current PDAG.

Returns:

Name	Type	Description
int	int	Number of directed edges

num_nodes property

Number of nodes in current PDAG.

Returns:

Name	Type	Description
int	int	Number of nodes

num_undir_edges property

Number of undirected edges in current PDAG.

Returns:

Name	Type	Description
int	int	Number of undirected edges

undir_edges property

Gives all undirected edges in current PDAG.

Returns:

Type	Description
list[tuple[str, str]]	list[tuple[str,str]]: List of undirected edges, sorted for determinism.

__init__(nodes=None, dir_edges=None, undir_edges=None)

PDAG constructor.

Parameters:

Name	Type	Description	Default
nodes	list[str] | None	Nodes in the PDAG. Defaults to None.	None
dir_edges	list[tuple[str, str]] | None	directed edges. Defaults to None.	None
undir_edges	list[tuple[str, str]] | None	undirected edges. Defaults to None.	None

Source code in mixpc/graphs.py

def __init__(
    self,
    nodes: list[str] | None = None,
    dir_edges: list[tuple[str, str]] | None = None,
    undir_edges: list[tuple[str, str]] | None = None,
) -> None:
    """PDAG constructor.

    Args:
        nodes (list[str] | None, optional): Nodes in the PDAG. Defaults to None.
        dir_edges (list[tuple[str,str]] | None, optional): directed edges. Defaults to None.
        undir_edges (list[tuple[str,str]] | None, optional): undirected edges. Defaults to None.
    """
    if nodes is None:
        nodes = []
    if dir_edges is None:
        dir_edges = []
    if undir_edges is None:
        undir_edges = []

    self._nodes = set(nodes)
    self._undir_edges: set[tuple[str, str]] = set()
    self._dir_edges: set[tuple[str, str]] = set()
    self._parents: defaultdict[str, set[str]] = defaultdict(set)
    self._children: defaultdict[str, set[str]] = defaultdict(set)
    self._neighbors: defaultdict[str, set[str]] = defaultdict(set)
    self._undirected_neighbors: defaultdict[str, set[str]] = defaultdict(set)

    for dir_edge in dir_edges:
        self._add_dir_edge(*dir_edge)
    for unir_edge in undir_edges:
        self._add_undir_edge(*unir_edge)

children(node)

Gives all children of node node.

Parameters:

Name	Type	Description	Default
node	str	node in current PDAG.	required

Returns:

Name	Type	Description
set	set[str]	set of children.

Source code in mixpc/graphs.py

def children(self, node: str) -> set[str]:
    """Gives all children of node `node`.

    Args:
        node (str): node in current PDAG.

    Returns:
        set: set of children.
    """
    if node in self._children:
        return self._children[node]
    else:
        return set()

copy()

Return a copy of the graph.

Source code in mixpc/graphs.py

def copy(self) -> PDAG:
    """Return a copy of the graph."""
    return PDAG(
        nodes=list(self._nodes),
        dir_edges=list(self._dir_edges),
        undir_edges=list(self._undir_edges),
    )

from_pandas_adjacency(pd_amat) classmethod

Build PDAG from a Pandas adjacency matrix.

Parameters:

Name	Type	Description	Default
pd_amat	DataFrame	input adjacency matrix.	required

Returns:

Type	Description
PDAG	PDAG

Source code in mixpc/graphs.py

@classmethod
def from_pandas_adjacency(cls, pd_amat: pd.DataFrame) -> PDAG:
    """Build PDAG from a Pandas adjacency matrix.

    Args:
        pd_amat (pd.DataFrame): input adjacency matrix.

    Returns:
        PDAG
    """
    assert pd_amat.shape[0] == pd_amat.shape[1]
    nodes = list(pd_amat.columns)

    all_connections = []
    start, end = np.where(pd_amat != 0)
    for idx, _ in enumerate(start):
        all_connections.append((pd_amat.columns[start[idx]], pd_amat.columns[end[idx]]))

    temp = [set(i) for i in all_connections]
    temp2 = [arc for arc in all_connections if temp.count(set(arc)) > 1]
    undir_edges = [tuple(item) for item in set(frozenset(item) for item in temp2)]

    dir_edges = [edge for edge in all_connections if edge not in temp2]

    return PDAG(nodes=nodes, dir_edges=dir_edges, undir_edges=undir_edges)

is_adjacent(i, j)

Return True if the graph contains an directed or undirected edge between i and j.

Parameters:

Name	Type	Description	Default
i	str	node i.	required
j	str	node j.	required

Returns:

Name	Type	Description
bool	bool	True if i-j or i->j or i<-j

Source code in mixpc/graphs.py

def is_adjacent(self, i: str, j: str) -> bool:
    """Return True if the graph contains an directed or undirected edge between i and j.

    Args:
        i (str): node i.
        j (str): node j.

    Returns:
        bool: True if i-j or i->j or i<-j
    """
    return (
        (i, j) in self._dir_edges
        or (j, i) in self._dir_edges
        or (i, j) in self._undir_edges
        or (j, i) in self._undir_edges
    )

is_clique(potential_clique)

Check every pair of node X potential_clique is adjacent.

Source code in mixpc/graphs.py

def is_clique(self, potential_clique: set[str]) -> bool:
    """Check every pair of node X potential_clique is adjacent."""
    return all(self.is_adjacent(i, j) for i, j in combinations(potential_clique, 2))

neighbors(node)

Gives all neighbors of node node.

Parameters:

Name	Type	Description	Default
node	str	node in current PDAG.	required

Returns:

Name	Type	Description
set	set[str]	set of neighbors.

Source code in mixpc/graphs.py

def neighbors(self, node: str) -> set[str]:
    """Gives all neighbors of node `node`.

    Args:
        node (str): node in current PDAG.

    Returns:
        set: set of neighbors.
    """
    if node in self._neighbors:
        return self._neighbors[node]
    else:
        return set()

parents(node)

Gives all parents of node node.

Parameters:

Name	Type	Description	Default
node	str	node in current PDAG.	required

Returns:

Name	Type	Description
set	set[str]	set of parents.

Source code in mixpc/graphs.py

def parents(self, node: str) -> set[str]:
    """Gives all parents of node `node`.

    Args:
        node (str): node in current PDAG.

    Returns:
        set: set of parents.
    """
    if node in self._parents:
        return self._parents[node]
    else:
        return set()

remove_edge(i, j)

Removes edge in question.

Parameters:

Name	Type	Description	Default
i	str	tail	required
j	str	head	required

Raises:

Type	Description
AssertionError	if edge does not exist

Source code in mixpc/graphs.py

def remove_edge(self, i: str, j: str) -> None:
    """Removes edge in question.

    Args:
        i (str): tail
        j (str): head

    Raises:
        AssertionError: if edge does not exist
    """
    if (i, j) not in self._dir_edges and (i, j) not in self._undir_edges:
        raise AssertionError("Edge does not exist in current PDAG")

    self._undir_edges.discard((i, j))
    self._dir_edges.discard((i, j))
    self._children[i].discard(j)
    self._parents[j].discard(i)
    self._neighbors[i].discard(j)
    self._neighbors[j].discard(i)
    self._undirected_neighbors[i].discard(j)
    self._undirected_neighbors[j].discard(i)

remove_node(node)

Remove a node from the graph.

Parameters:

Name	Type	Description	Default
node	str	node to remove	required

Source code in mixpc/graphs.py

def remove_node(self, node: str) -> None:
    """Remove a node from the graph.

    Args:
        node (str): node to remove
    """
    self._nodes.remove(node)

    self._dir_edges = {(i, j) for i, j in self._dir_edges if node not in {i, j}}

    self._undir_edges = {(i, j) for i, j in self._undir_edges if node not in {i, j}}

    for child in self._children[node]:
        self._parents[child].remove(node)
        self._neighbors[child].remove(node)

    for parent in self._parents[node]:
        self._children[parent].remove(node)
        self._neighbors[parent].remove(node)

    for u_nbr in self._undirected_neighbors[node]:
        self._undirected_neighbors[u_nbr].remove(node)
        self._neighbors[u_nbr].remove(node)

    self._parents.pop(node, "I was never here")
    self._children.pop(node, "I was never here")
    self._neighbors.pop(node, "I was never here")
    self._undirected_neighbors.pop(node, "I was never here")

show()

Plot PDAG.

Source code in mixpc/graphs.py

def show(self) -> None:
    """Plot PDAG."""
    graph = self.to_networkx()
    pos = nx.circular_layout(graph)
    nx.draw(graph, pos=pos, with_labels=True)

to_allDAGs()

Recursion algorithm which recursively applies the following steps.

1. Orient the first undirected edge found.
2. Apply Meek rules.
3. Recurse with each direction of the oriented edge.

This corresponds to Algorithm 2 in Wienöbst et al. (2023).

References

Wienöbst, Marcel, et al. "Efficient enumeration of Markov equivalent DAGs." Proceedings of the AAAI Conference on Artificial Intelligence. Vol. 37. No. 10. 2023.

Source code in mixpc/graphs.py

def to_allDAGs(self) -> list[DAG]:  # noqa: N802
    """Recursion algorithm which recursively applies the following steps.

        1. Orient the first undirected edge found.
        2. Apply Meek rules.
        3. Recurse with each direction of the oriented edge.
    This corresponds to Algorithm 2 in Wienöbst et al. (2023).

    References:
        Wienöbst, Marcel, et al. "Efficient enumeration of Markov equivalent DAGs."
        Proceedings of the AAAI Conference on Artificial Intelligence.
        Vol. 37. No. 10. 2023.
    """
    all_dags: list[DAG] = []
    self._meek_mec_enumeration(pdag=self, dag_list=all_dags)
    return all_dags

to_dag()

Algorithm as described in Chickering (2002).

1. From PDAG P create DAG G containing all directed edges from P
2. Repeat the following: Select node v in P s.t.
    i. v has no outgoing edges (children) i.e. \\(ch(v) = \\emptyset \\)

    ii. \\(neigh(v) \\neq \\emptyset\\)
        Then \\( (pa(v) \\cup (neigh(v) \\) form a clique.
        For each v that is in a clique and is part of an undirected edge in P
        i.e. w - v, insert a directed edge w -> v in G.
        Remove v and all incident edges from P and continue with next node.
        Until all nodes have been deleted from P.

Returns:

Type	Description
DiGraph	nx.DiGraph: DAG that belongs to the MEC implied by the PDAG

Source code in mixpc/graphs.py

def to_dag(self) -> nx.DiGraph:
    r"""Algorithm as described in Chickering (2002).

        1. From PDAG P create DAG G containing all directed edges from P
        2. Repeat the following: Select node v in P s.t.
            i. v has no outgoing edges (children) i.e. \\(ch(v) = \\emptyset \\)

            ii. \\(neigh(v) \\neq \\emptyset\\)
                Then \\( (pa(v) \\cup (neigh(v) \\) form a clique.
                For each v that is in a clique and is part of an undirected edge in P
                i.e. w - v, insert a directed edge w -> v in G.
                Remove v and all incident edges from P and continue with next node.
                Until all nodes have been deleted from P.

    Returns:
        nx.DiGraph: DAG that belongs to the MEC implied by the PDAG
    """
    pdag = self.copy()

    dag = nx.DiGraph()
    dag.add_nodes_from(pdag.nodes)
    dag.add_edges_from(pdag.dir_edges)

    if pdag.num_undir_edges == 0:
        return dag
    else:
        while pdag.num_nodes > 0:
            # find node with (1) no directed outgoing edges and
            #                (2) the set of undirected neighbors is either empty or
            #                    undirected neighbors + parents of X are a clique
            found = False
            for node in pdag.nodes:
                children = pdag.children(node)
                neighbors = pdag.neighbors(node)
                # pdag._undirected_neighbors[node]
                parents = pdag.parents(node)
                potential_clique_members = neighbors.union(parents)

                is_clique = pdag.is_clique(potential_clique_members)

                if not children and (not neighbors or is_clique):
                    found = True
                    # add all edges of node as outgoing edges to dag
                    for edge in pdag.undir_edges:
                        if node in edge:
                            incident_node = next(iter(set(edge) - {node}))
                            dag.add_edge(incident_node, node)

                    pdag.remove_node(node)
                    break

            if not found:
                logger.warning("PDAG not extendible: Random DAG on skeleton drawn.")

                dag = nx.from_pandas_adjacency(self._amat_to_dag(), create_using=nx.DiGraph)

                break

        return dag

to_networkx()

Convert to networkx graph.

Returns:

Type	Description
MultiDiGraph	nx.MultiDiGraph: Graph with directed and undirected edges.

Source code in mixpc/graphs.py

def to_networkx(self) -> nx.MultiDiGraph:
    """Convert to networkx graph.

    Returns:
        nx.MultiDiGraph: Graph with directed and undirected edges.
    """
    nx_pdag = nx.MultiDiGraph()
    nx_pdag.add_nodes_from(self.nodes)
    nx_pdag.add_edges_from(self.dir_edges)
    for edge in self.undir_edges:
        nx_pdag.add_edge(*edge)
        nx_pdag.add_edge(*edge[::-1])

    return nx_pdag

to_random_dag()

Provides a random DAG residing in the MEC.

Returns:

Type	Description
DAG	nx.DiGraph: random DAG living in MEC

Source code in mixpc/graphs.py

def to_random_dag(self) -> DAG:
    """Provides a random DAG residing in the MEC.

    Returns:
        nx.DiGraph: random DAG living in MEC
    """
    to_dag_candidate = self.copy()
    rng = np.random.default_rng()

    while to_dag_candidate.num_undir_edges > 0:
        chosen_edge = to_dag_candidate.undir_edges[rng.choice(to_dag_candidate.num_undir_edges)]
        choose_orientation = [chosen_edge, chosen_edge[::-1]]
        node_i, node_j = choose_orientation[rng.choice(len(choose_orientation))]

        to_dag_candidate.undir_to_dir_edge(tail=node_i, head=node_j)
        to_dag_candidate = to_dag_candidate._apply_meek_rules(pdag=to_dag_candidate)

    return DAG.from_pandas_adjacency(to_dag_candidate.adjacency_matrix)

undir_neighbors(node)

Gives all undirected neighbors of node node.

Parameters:

Name	Type	Description	Default
node	str	node in current PDAG.	required

Returns:

Name	Type	Description
set	set[str]	set of undirected neighbors.

Source code in mixpc/graphs.py

def undir_neighbors(self, node: str) -> set[str]:
    """Gives all undirected neighbors of node `node`.

    Args:
        node (str): node in current PDAG.

    Returns:
        set: set of undirected neighbors.
    """
    if node in self._undirected_neighbors:
        return self._undirected_neighbors[node]
    else:
        return set()

undir_to_dir_edge(tail, head)

Takes a undirected edge and turns it into a directed one.

tail indicates the starting node of the edge and head the end node, i.e. tail -> head.

Parameters:

Name	Type	Description	Default
tail	str	starting node	required
head	str	end node	required

Raises:

Type	Description
AssertionError	if edge does not exist or is not undirected.

Source code in mixpc/graphs.py

def undir_to_dir_edge(self, tail: str, head: str) -> None:
    """Takes a undirected edge and turns it into a directed one.

    tail indicates the starting node of the edge and head the end node, i.e.
    tail -> head.

    Args:
        tail (str): starting node
        head (str): end node

    Raises:
        AssertionError: if edge does not exist or is not undirected.
    """
    if (tail, head) not in self._undir_edges and (head, tail) not in self._undir_edges:
        raise AssertionError("Edge seems not to be undirected or even there at all.")
    self._undir_edges.discard((tail, head))
    self._undir_edges.discard((head, tail))
    self._neighbors[tail].discard(head)
    self._neighbors[head].discard(tail)
    self._undirected_neighbors[tail].discard(head)
    self._undirected_neighbors[head].discard(tail)

    self._add_dir_edge(i=tail, j=head)

vstructs()

Retrieve v-structures.

Returns:

Name	Type	Description
set	set[tuple[str, str]]	set of all v-structures

Source code in mixpc/graphs.py

def vstructs(self) -> set[tuple[str, str]]:
    """Retrieve v-structures.

    Returns:
        set: set of all v-structures
    """
    vstructures = set()
    for node in self._nodes:
        for p1, p2 in combinations(self._parents[node], 2):
            if p1 not in self._parents[p2] and p2 not in self._parents[p1]:
                vstructures.add((p1, node))
                vstructures.add((p2, node))
    return vstructures

Bases: GRAPH

General class for dealing with directed acyclic graph i.e.

graphs that are directed and must not contain any cycles.

Source code in mixpc/graphs.py

class DAG(GRAPH):
    """General class for dealing with directed acyclic graph i.e.

    graphs that are directed and must not contain any cycles.
    """

    def __init__(
        self,
        nodes: list[str] | None = None,
        edges: list[tuple[str, str]] | None = None,
    ) -> None:
        """DAG constructor.

        Args:
            nodes (list[str] | None, optional): Nodes. Defaults to None.
            edges (list[tuple[str,str]] | None, optional): Edges. Defaults to None.
        """
        if nodes is None:
            nodes = []
        if edges is None:
            edges = []

        self._nodes: set[str] = set(nodes)
        self._edges: set[tuple[str, str]] = set()
        self._parents: defaultdict[str, set[str]] = defaultdict(set)
        self._children: defaultdict[str, set[str]] = defaultdict(set)
        self._random_state: np.random.Generator = np.random.default_rng(seed=2023)

        for edge in edges:
            self._add_edge(*edge)

    def _add_node(self, node: str) -> None:
        self._nodes.add(node)

    def _add_edge(self, i: str, j: str) -> None:
        self._nodes.add(i)
        self._nodes.add(j)
        self._edges.add((i, j))

        # Check if graph is acyclic
        if not self.is_acyclic():
            raise ValueError(
                "The edge set you provided \
                induces one or more cycles.\
                Check your input!"
            )

        self._children[i].add(j)
        self._parents[j].add(i)

    @property
    def random_state(self) -> np.random.Generator:
        """Current random state.

        Returns:
            np.random.Generator: Generator object.
        """
        return self._random_state

    @random_state.setter
    def random_state(self, r: np.random.Generator) -> None:
        if not isinstance(r, np.random.Generator):
            raise AssertionError("Specify numpy random number generator object!")
        self._random_state = r

    def add_edge(self, edge: tuple[str, str]) -> None:
        """Add edge to DAG.

        Args:
            edge (tuple[str, str]): Edge to add
        """
        self._add_edge(*edge)

    def add_node(self, node: str) -> None:
        """Add node to DAG.

        Args:
            node (str): node to add
        """
        self._add_node(node)

    def add_edges_from(self, edges: list[tuple[str, str]]) -> None:
        """Add multiple edges to DAG.

        Args:
            edges (list[tuple[str, str]]): Edges to add
        """
        for edge in edges:
            self.add_edge(edge=edge)

    def add_nodes_from(self, nodes: list[str]) -> None:
        """Add multiple nodes to DAG.

        Args:
            nodes (list[str]): nodes to add
        """
        for node in nodes:
            self.add_node(node)

    def children(self, of_node: str) -> list[str]:
        """Gives all children of node `node`.

        Args:
            of_node (str): node in current DAG.

        Returns:
            list: of children.
        """
        if of_node in self._children:
            return list(self._children[of_node])
        else:
            return []

    def parents(self, of_node: str) -> list[str]:
        """Gives all parents of node `node`.

        Args:
            of_node (str): node in current DAG.

        Returns:
            list: of parents.
        """
        if of_node in self._parents:
            return list(self._parents[of_node])
        else:
            return []

    def induced_subgraph(self, nodes: list[str]) -> DAG:
        """Returns the induced subgraph on the nodes in `nodes`.

        Args:
            nodes (list[str]): List of nodes.

        Returns:
            DAG: Induced subgraph.
        """
        edges = [(i, j) for i, j in self.edges if i in nodes and j in nodes]
        return DAG(nodes=nodes, edges=edges)

    def is_adjacent(self, i: str, j: str) -> bool:
        """Return True if the graph contains an directed edge between i and j.

        Args:
            i (str): node i.
            j (str): node j.

        Returns:
            bool: True if i->j or i<-j
        """
        return (j, i) in self.edges or (i, j) in self.edges

    def is_clique(self, potential_clique: set[str]) -> bool:
        """Check every pair of node X potential_clique is adjacent."""
        return all(self.is_adjacent(i, j) for i, j in combinations(potential_clique, 2))

    def is_acyclic(self) -> bool:
        """Check if the graph is acyclic.

        Returns:
            bool: True if graph is acyclic.
        """
        nx_dag = self.to_networkx()
        acyclic: bool = nx.is_directed_acyclic_graph(nx_dag)
        return acyclic

    @classmethod
    def from_pandas_adjacency(cls, pd_amat: pd.DataFrame, *args: Any, **kwargs: Any) -> DAG:
        """Build DAG from a Pandas adjacency matrix.

        Args:
            pd_amat (pd.DataFrame): input adjacency matrix.
            args (Any): Additional arguments.
            kwargs (Any): Additional arguments.

        Returns:
            DAG
        """
        assert pd_amat.shape[0] == pd_amat.shape[1]
        nodes = list(pd_amat.columns)

        all_connections = []
        start, end = np.where(pd_amat != 0)
        for idx, _ in enumerate(start):
            all_connections.append((pd_amat.columns[start[idx]], pd_amat.columns[end[idx]]))

        temp = [set(i) for i in all_connections]
        temp2 = [arc for arc in all_connections if temp.count(set(arc)) > 1]

        dir_edges = [edge for edge in all_connections if edge not in temp2]

        return DAG(nodes=nodes, edges=dir_edges)

    def remove_edge(self, i: str, j: str) -> None:
        """Removes edge in question.

        Args:
            i (str): tail
            j (str): head

        Raises:
            AssertionError: if edge does not exist
        """
        if (i, j) not in self.edges:
            raise AssertionError("Edge does not exist in current DAG")

        self._edges.discard((i, j))
        self._children[i].discard(j)
        self._parents[j].discard(i)

    def remove_node(self, node: str) -> None:
        """Remove a node from the graph."""
        self._nodes.remove(node)

        self._edges = {(i, j) for i, j in self._edges if node not in {i, j}}

        for child in self._children[node]:
            self._parents[child].remove(node)

        for parent in self._parents[node]:
            self._children[parent].remove(node)

        self._parents.pop(node, "I was never here")
        self._children.pop(node, "I was never here")

    @property
    def adjacency_matrix(self) -> pd.DataFrame:
        """Returns adjacency matrix.

        The i,jth entry being one indicates that there is an edge
        from i to j. A zero indicates that there is no edge.

        Returns:
            pd.DataFrame: adjacency matrix
        """
        amat = pd.DataFrame(
            np.zeros([self.num_nodes, self.num_nodes]),
            index=self.nodes,
            columns=self.nodes,
        )
        for edge in self.edges:
            amat.loc[edge] = 1
        return amat

    def vstructs(self) -> set[tuple[str, str]]:
        """Retrieve v-structures.

        Returns:
            set: set of all v-structures
        """
        vstructures = set()
        for node in self._nodes:
            for p1, p2 in combinations(self._parents[node], 2):
                if p1 not in self._parents[p2] and p2 not in self._parents[p1]:
                    vstructures.add((p1, node))
                    vstructures.add((p2, node))
        return vstructures

    def copy(self) -> DAG:
        """Return a copy of the graph."""
        return DAG(nodes=list(self._nodes), edges=list(self._edges))

    def show(self) -> None:
        """Plot DAG."""
        graph = self.to_networkx()
        pos = nx.circular_layout(graph)
        nx.draw(graph, pos=pos, with_labels=True)

    def to_networkx(self) -> nx.DiGraph:
        """Convert to networkx graph.

        Returns:
            nx.MultiDiGraph: Graph with directed and undirected edges.
        """
        nx_dag = nx.DiGraph()
        nx_dag.add_nodes_from(self.nodes)
        nx_dag.add_edges_from(self.edges)

        return nx_dag

    @property
    def nodes(self) -> list[str]:
        """Get all nods in current DAG.

        Returns:
            list: list of nodes.
        """
        return sorted(list(self._nodes))

    @property
    def num_nodes(self) -> int:
        """Number of nodes in current DAG.

        Returns:
            int: Number of nodes
        """
        return len(self._nodes)

    @property
    def num_edges(self) -> int:
        """Number of directed edges in current DAG.

        Returns:
            int: Number of directed edges
        """
        return len(self._edges)

    @property
    def sparsity(self) -> float:
        """Sparsity of the graph.

        Returns:
            float: in [0,1]
        """
        s = self.num_nodes
        return self.num_edges / s / (s - 1) * 2

    @property
    def edges(self) -> list[tuple[str, str]]:
        """Gives all directed edges in current DAG.

        Returns:
            list[tuple[str,str]]: List of directed edges.
        """
        return list(self._edges)

    @property
    def causal_order(self) -> list[str]:
        """Returns the causal order of the current graph.

        Note that this order is in general not unique.

        Returns:
            list[str]: Causal order
        """
        return list(nx.lexicographical_topological_sort(self.to_networkx()))

    @property
    def sink_nodes(self) -> list[str]:
        """Returns all sink nodes, i.e.

        nodes with no descendents in particular no children.

        Returns:
            list[str]: list of sink nodes.
        """
        return [s for b, s in zip([self.children(of_node=node) == [] for node in self.nodes], self.nodes) if b]

    @property
    def source_nodes(self) -> list[str]:
        """Returns all source nodes, i.e.

        nodes with no ancesters in particular no parents.

        Returns:
            list[str]: list of sink nodes.
        """
        return [s for b, s in zip([self.parents(of_node=node) == [] for node in self.nodes], self.nodes) if b]

    @property
    def max_in_degree(self) -> int:
        """Maximum in-degree of the graph.

        Returns:
            int: Maximum in-degree
        """
        return max(len(self._parents[node]) for node in self._nodes)

    @property
    def max_out_degree(self) -> int:
        """Maximum out-degree of the graph.

        Returns:
            int: Maximum out-degree
        """
        return max(len(self._children[node]) for node in self._nodes)

    @classmethod
    def from_nx(cls, nx_dag: nx.DiGraph, *args: Any, **kwargs: Any) -> DAG:
        """Convert to DAG from nx.DiGraph.

        Args:
            nx_dag (nx.DiGraph): DAG in question.
            args (Any): additional arguments
            kwargs (Any): additional arguments

        Returns:
            DAG

        Raises:
            TypeError: If DAG is not nx.DiGraph
        """
        if not isinstance(nx_dag, nx.DiGraph):
            raise TypeError("DAG must be of type nx.DiGraph")
        return DAG(nodes=list(nx_dag.nodes), edges=list(nx_dag.edges))

    def to_cpdag(self) -> PDAG:
        """Convert DAG to CPDAG.

        Returns:
            PDAG: CPDAG representing the MEC.
        """
        return dag2cpdag(dag=self.to_networkx())

adjacency_matrix property

Returns adjacency matrix.

The i,jth entry being one indicates that there is an edge from i to j. A zero indicates that there is no edge.

Returns:

Type	Description
DataFrame	pd.DataFrame: adjacency matrix

causal_order property

Returns the causal order of the current graph.

Note that this order is in general not unique.

Returns:

Type	Description
list[str]	list[str]: Causal order

edges property

Gives all directed edges in current DAG.

Returns:

Type	Description
list[tuple[str, str]]	list[tuple[str,str]]: List of directed edges.

max_in_degree property

Maximum in-degree of the graph.

Returns:

Name	Type	Description
int	int	Maximum in-degree

max_out_degree property

Maximum out-degree of the graph.

Returns:

Name	Type	Description
int	int	Maximum out-degree

nodes property

Get all nods in current DAG.

Returns:

Name	Type	Description
list	list[str]	list of nodes.

num_edges property

Number of directed edges in current DAG.

Returns:

Name	Type	Description
int	int	Number of directed edges

num_nodes property

Number of nodes in current DAG.

Returns:

Name	Type	Description
int	int	Number of nodes

random_state property writable

Current random state.

Returns:

Type	Description
Generator	np.random.Generator: Generator object.

sink_nodes property

Returns all sink nodes, i.e.

nodes with no descendents in particular no children.

Returns:

Type	Description
list[str]	list[str]: list of sink nodes.

source_nodes property

Returns all source nodes, i.e.

nodes with no ancesters in particular no parents.

Returns:

Type	Description
list[str]	list[str]: list of sink nodes.

sparsity property

Sparsity of the graph.

Returns:

Name	Type	Description
float	float	in [0,1]

__init__(nodes=None, edges=None)

DAG constructor.

Parameters:

Name	Type	Description	Default
nodes	list[str] | None	Nodes. Defaults to None.	None
edges	list[tuple[str, str]] | None	Edges. Defaults to None.	None

Source code in mixpc/graphs.py

def __init__(
    self,
    nodes: list[str] | None = None,
    edges: list[tuple[str, str]] | None = None,
) -> None:
    """DAG constructor.

    Args:
        nodes (list[str] | None, optional): Nodes. Defaults to None.
        edges (list[tuple[str,str]] | None, optional): Edges. Defaults to None.
    """
    if nodes is None:
        nodes = []
    if edges is None:
        edges = []

    self._nodes: set[str] = set(nodes)
    self._edges: set[tuple[str, str]] = set()
    self._parents: defaultdict[str, set[str]] = defaultdict(set)
    self._children: defaultdict[str, set[str]] = defaultdict(set)
    self._random_state: np.random.Generator = np.random.default_rng(seed=2023)

    for edge in edges:
        self._add_edge(*edge)

add_edge(edge)

Add edge to DAG.

Parameters:

Name	Type	Description	Default
edge	tuple[str, str]	Edge to add	required

Source code in mixpc/graphs.py

def add_edge(self, edge: tuple[str, str]) -> None:
    """Add edge to DAG.

    Args:
        edge (tuple[str, str]): Edge to add
    """
    self._add_edge(*edge)

add_edges_from(edges)

Add multiple edges to DAG.

Parameters:

Name	Type	Description	Default
edges	list[tuple[str, str]]	Edges to add	required

Source code in mixpc/graphs.py

def add_edges_from(self, edges: list[tuple[str, str]]) -> None:
    """Add multiple edges to DAG.

    Args:
        edges (list[tuple[str, str]]): Edges to add
    """
    for edge in edges:
        self.add_edge(edge=edge)

add_node(node)

Add node to DAG.

Parameters:

Name	Type	Description	Default
node	str	node to add	required

Source code in mixpc/graphs.py

def add_node(self, node: str) -> None:
    """Add node to DAG.

    Args:
        node (str): node to add
    """
    self._add_node(node)

add_nodes_from(nodes)

Add multiple nodes to DAG.

Parameters:

Name	Type	Description	Default
nodes	list[str]	nodes to add	required

Source code in mixpc/graphs.py

def add_nodes_from(self, nodes: list[str]) -> None:
    """Add multiple nodes to DAG.

    Args:
        nodes (list[str]): nodes to add
    """
    for node in nodes:
        self.add_node(node)

children(of_node)

Gives all children of node node.

Parameters:

Name	Type	Description	Default
of_node	str	node in current DAG.	required

Returns:

Name	Type	Description
list	list[str]	of children.

Source code in mixpc/graphs.py

def children(self, of_node: str) -> list[str]:
    """Gives all children of node `node`.

    Args:
        of_node (str): node in current DAG.

    Returns:
        list: of children.
    """
    if of_node in self._children:
        return list(self._children[of_node])
    else:
        return []

copy()

Return a copy of the graph.

Source code in mixpc/graphs.py

def copy(self) -> DAG:
    """Return a copy of the graph."""
    return DAG(nodes=list(self._nodes), edges=list(self._edges))

from_nx(nx_dag, *args, **kwargs) classmethod

Convert to DAG from nx.DiGraph.

Parameters:

Name	Type	Description	Default
nx_dag	DiGraph	DAG in question.	required
args	Any	additional arguments	()
kwargs	Any	additional arguments	{}

Returns:

Type	Description
DAG	DAG

Raises:

Type	Description
TypeError	If DAG is not nx.DiGraph

Source code in mixpc/graphs.py

@classmethod
def from_nx(cls, nx_dag: nx.DiGraph, *args: Any, **kwargs: Any) -> DAG:
    """Convert to DAG from nx.DiGraph.

    Args:
        nx_dag (nx.DiGraph): DAG in question.
        args (Any): additional arguments
        kwargs (Any): additional arguments

    Returns:
        DAG

    Raises:
        TypeError: If DAG is not nx.DiGraph
    """
    if not isinstance(nx_dag, nx.DiGraph):
        raise TypeError("DAG must be of type nx.DiGraph")
    return DAG(nodes=list(nx_dag.nodes), edges=list(nx_dag.edges))

from_pandas_adjacency(pd_amat, *args, **kwargs) classmethod

Build DAG from a Pandas adjacency matrix.

Parameters:

Name	Type	Description	Default
pd_amat	DataFrame	input adjacency matrix.	required
args	Any	Additional arguments.	()
kwargs	Any	Additional arguments.	{}

Returns:

Type	Description
DAG	DAG

Source code in mixpc/graphs.py

@classmethod
def from_pandas_adjacency(cls, pd_amat: pd.DataFrame, *args: Any, **kwargs: Any) -> DAG:
    """Build DAG from a Pandas adjacency matrix.

    Args:
        pd_amat (pd.DataFrame): input adjacency matrix.
        args (Any): Additional arguments.
        kwargs (Any): Additional arguments.

    Returns:
        DAG
    """
    assert pd_amat.shape[0] == pd_amat.shape[1]
    nodes = list(pd_amat.columns)

    all_connections = []
    start, end = np.where(pd_amat != 0)
    for idx, _ in enumerate(start):
        all_connections.append((pd_amat.columns[start[idx]], pd_amat.columns[end[idx]]))

    temp = [set(i) for i in all_connections]
    temp2 = [arc for arc in all_connections if temp.count(set(arc)) > 1]

    dir_edges = [edge for edge in all_connections if edge not in temp2]

    return DAG(nodes=nodes, edges=dir_edges)

induced_subgraph(nodes)

Returns the induced subgraph on the nodes in nodes.

Parameters:

Name	Type	Description	Default
nodes	list[str]	List of nodes.	required

Returns:

Name	Type	Description
DAG	DAG	Induced subgraph.

Source code in mixpc/graphs.py

def induced_subgraph(self, nodes: list[str]) -> DAG:
    """Returns the induced subgraph on the nodes in `nodes`.

    Args:
        nodes (list[str]): List of nodes.

    Returns:
        DAG: Induced subgraph.
    """
    edges = [(i, j) for i, j in self.edges if i in nodes and j in nodes]
    return DAG(nodes=nodes, edges=edges)

is_acyclic()

Check if the graph is acyclic.

Returns:

Name	Type	Description
bool	bool	True if graph is acyclic.

Source code in mixpc/graphs.py

def is_acyclic(self) -> bool:
    """Check if the graph is acyclic.

    Returns:
        bool: True if graph is acyclic.
    """
    nx_dag = self.to_networkx()
    acyclic: bool = nx.is_directed_acyclic_graph(nx_dag)
    return acyclic

is_adjacent(i, j)

Return True if the graph contains an directed edge between i and j.

Parameters:

Name	Type	Description	Default
i	str	node i.	required
j	str	node j.	required

Returns:

Name	Type	Description
bool	bool	True if i->j or i<-j

Source code in mixpc/graphs.py

def is_adjacent(self, i: str, j: str) -> bool:
    """Return True if the graph contains an directed edge between i and j.

    Args:
        i (str): node i.
        j (str): node j.

    Returns:
        bool: True if i->j or i<-j
    """
    return (j, i) in self.edges or (i, j) in self.edges

is_clique(potential_clique)

Check every pair of node X potential_clique is adjacent.

Source code in mixpc/graphs.py

def is_clique(self, potential_clique: set[str]) -> bool:
    """Check every pair of node X potential_clique is adjacent."""
    return all(self.is_adjacent(i, j) for i, j in combinations(potential_clique, 2))

parents(of_node)

Gives all parents of node node.

Parameters:

Name	Type	Description	Default
of_node	str	node in current DAG.	required

Returns:

Name	Type	Description
list	list[str]	of parents.

Source code in mixpc/graphs.py

def parents(self, of_node: str) -> list[str]:
    """Gives all parents of node `node`.

    Args:
        of_node (str): node in current DAG.

    Returns:
        list: of parents.
    """
    if of_node in self._parents:
        return list(self._parents[of_node])
    else:
        return []

remove_edge(i, j)

Removes edge in question.

Parameters:

Name	Type	Description	Default
i	str	tail	required
j	str	head	required

Raises:

Type	Description
AssertionError	if edge does not exist

Source code in mixpc/graphs.py

def remove_edge(self, i: str, j: str) -> None:
    """Removes edge in question.

    Args:
        i (str): tail
        j (str): head

    Raises:
        AssertionError: if edge does not exist
    """
    if (i, j) not in self.edges:
        raise AssertionError("Edge does not exist in current DAG")

    self._edges.discard((i, j))
    self._children[i].discard(j)
    self._parents[j].discard(i)

remove_node(node)

Remove a node from the graph.

Source code in mixpc/graphs.py

def remove_node(self, node: str) -> None:
    """Remove a node from the graph."""
    self._nodes.remove(node)

    self._edges = {(i, j) for i, j in self._edges if node not in {i, j}}

    for child in self._children[node]:
        self._parents[child].remove(node)

    for parent in self._parents[node]:
        self._children[parent].remove(node)

    self._parents.pop(node, "I was never here")
    self._children.pop(node, "I was never here")

show()

Plot DAG.

Source code in mixpc/graphs.py

def show(self) -> None:
    """Plot DAG."""
    graph = self.to_networkx()
    pos = nx.circular_layout(graph)
    nx.draw(graph, pos=pos, with_labels=True)

to_cpdag()

Convert DAG to CPDAG.

Returns:

Name	Type	Description
PDAG	PDAG	CPDAG representing the MEC.

Source code in mixpc/graphs.py

def to_cpdag(self) -> PDAG:
    """Convert DAG to CPDAG.

    Returns:
        PDAG: CPDAG representing the MEC.
    """
    return dag2cpdag(dag=self.to_networkx())

to_networkx()

Convert to networkx graph.

Returns:

Type	Description
DiGraph	nx.MultiDiGraph: Graph with directed and undirected edges.

Source code in mixpc/graphs.py

def to_networkx(self) -> nx.DiGraph:
    """Convert to networkx graph.

    Returns:
        nx.MultiDiGraph: Graph with directed and undirected edges.
    """
    nx_dag = nx.DiGraph()
    nx_dag.add_nodes_from(self.nodes)
    nx_dag.add_edges_from(self.edges)

    return nx_dag

vstructs()

Retrieve v-structures.

Returns:

Name	Type	Description
set	set[tuple[str, str]]	set of all v-structures

Source code in mixpc/graphs.py

def vstructs(self) -> set[tuple[str, str]]:
    """Retrieve v-structures.

    Returns:
        set: set of all v-structures
    """
    vstructures = set()
    for node in self._nodes:
        for p1, p2 in combinations(self._parents[node], 2):
            if p1 not in self._parents[p2] and p2 not in self._parents[p1]:
                vstructures.add((p1, node))
                vstructures.add((p2, node))
    return vstructures

Bases: GRAPH

Class for dealing with undirected graph i.e. graphs that only contain undirected edges.

Source code in mixpc/graphs.py

class UGRAPH(GRAPH):
    """Class for dealing with undirected graph i.e. graphs that only contain undirected edges."""

    def __init__(
        self,
        nodes: list[str] | None = None,
        edges: list[tuple[str, str]] | None = None,
    ) -> None:
        """UGRAPH constructor.

        Args:
            nodes (list[str] | None, optional): Nodes. Defaults to None.
            edges (list[tuple[str,str]] | None, optional): Edges. Defaults to None.
        """
        if nodes is None:
            nodes = []
        if edges is None:
            edges = []

        self._nodes: set[str] = set(nodes)
        self._edges: set[tuple[str, str]] = set()
        self._neighbors: defaultdict[str, set[str]] = defaultdict(set)

        for edge in edges:
            self._add_edge(*edge)

    def _add_edge(self, i: str, j: str) -> None:
        self._nodes.add(i)
        self._nodes.add(j)
        self._edges.add((i, j))

        self._neighbors[i].add(j)
        self._neighbors[j].add(i)

    def neighbors(self, node: str) -> set[str]:
        """Gives all neighbors of node `node`.

        Args:
            node (str): node in current UGRAPH.

        Returns:
            set: set of neighbors.
        """
        if node in self._neighbors:
            return self._neighbors[node]
        else:
            return set()

    def is_adjacent(self, i: str, j: str) -> bool:
        """Return True if the graph contains an undirected edge between i and j.

        Args:
            i (str): node i.
            j (str): node j.

        Returns:
            bool: True if i - j
        """
        return (i, j) in self._edges or (j, i) in self._edges

    def is_clique(self, potential_clique: set[str]) -> bool:
        """Check every pair of nodes in potential_clique is adjacent."""
        return all(self.is_adjacent(i, j) for i, j in combinations(potential_clique, 2))

    @classmethod
    def from_pandas_adjacency(cls, pd_amat: pd.DataFrame) -> UGRAPH:
        """Build UGRAPH from a Pandas adjacency matrix.

        Args:
            pd_amat (pd.DataFrame): input adjacency matrix.

        Returns:
            UGRAPH
        """
        assert pd_amat.shape[0] == pd_amat.shape[1]
        nodes = list(pd_amat.columns)

        all_connections = []
        start, end = np.where(pd_amat != 0)
        for idx, _ in enumerate(start):
            all_connections.append((pd_amat.columns[start[idx]], pd_amat.columns[end[idx]]))

        edges = [tuple(item) for item in set(frozenset(item) for item in all_connections)]

        return UGRAPH(nodes=nodes, edges=edges)

    def remove_edge(self, i: str, j: str) -> None:
        """Removes edge in question.

        Args:
            i (str): first node
            j (str): second node

        Raises:
            AssertionError: if edge does not exist
        """
        if not self.is_adjacent(i, j):
            raise AssertionError("Edge does not exist in current UGRAPH")

        self._edges.discard((i, j))
        self._edges.discard((j, i))
        self._neighbors[i].discard(j)
        self._neighbors[j].discard(i)

    def remove_node(self, node: str) -> None:
        """Remove a node from the graph.

        Args:
            node (str): node to remove
        """
        self._nodes.remove(node)

        self._edges = {(i, j) for i, j in self._edges if node not in {i, j}}

        for nbr in self._neighbors[node]:
            self._neighbors[nbr].discard(node)

        self._neighbors.pop(node, "I was never here")

    @property
    def adjacency_matrix(self) -> pd.DataFrame:
        """Returns adjacency matrix.

        The i,jth entry being one indicates that there is an undirected edge
        between i and j. A zero indicates that there is no edge. The matrix
        is symmetric.

        Returns:
            pd.DataFrame: adjacency matrix
        """
        amat = pd.DataFrame(
            np.zeros([self.num_nodes, self.num_nodes]),
            index=self.nodes,
            columns=self.nodes,
        )
        for edge in self.edges:
            amat.loc[edge] = amat.loc[edge[::-1]] = 1
        return amat

    @property
    def causal_order(self) -> None:
        """Causal order is None.

        This is because undirected graphs do not imply a causal order.

        Returns:
            None: None
        """
        return None

    def copy(self) -> UGRAPH:
        """Return a copy of the graph."""
        return UGRAPH(nodes=list(self._nodes), edges=list(self._edges))

    def show(self) -> None:
        """Plot UGRAPH."""
        graph = self.to_networkx()
        pos = nx.circular_layout(graph)
        nx.draw(graph, pos=pos, with_labels=True)

    def to_networkx(self) -> nx.Graph:
        """Convert to networkx graph.

        Returns:
            nx.Graph: Undirected networkx graph.
        """
        nx_ugraph = nx.Graph()
        nx_ugraph.add_nodes_from(self.nodes)
        nx_ugraph.add_edges_from(self.edges)
        return nx_ugraph

    @property
    def nodes(self) -> list[str]:
        """Get all nodes in current UGRAPH.

        Returns:
            list: list of nodes.
        """
        return sorted(list(self._nodes))

    @property
    def num_nodes(self) -> int:
        """Number of nodes in current UGRAPH.

        Returns:
            int: Number of nodes
        """
        return len(self._nodes)

    @property
    def num_edges(self) -> int:
        """Number of edges in current UGRAPH.

        Returns:
            int: Number of edges
        """
        return len(self._edges)

    @property
    def edges(self) -> list[tuple[str, str]]:
        """Gives all edges in current UGRAPH.

        Returns:
            list[tuple[str,str]]: List of edges.
        """
        return list(self._edges)

adjacency_matrix property

Returns adjacency matrix.

The i,jth entry being one indicates that there is an undirected edge between i and j. A zero indicates that there is no edge. The matrix is symmetric.

Returns:

Type	Description
DataFrame	pd.DataFrame: adjacency matrix

causal_order property

Causal order is None.

This is because undirected graphs do not imply a causal order.

Returns:

Name	Type	Description
None	None	None

edges property

Gives all edges in current UGRAPH.

Returns:

Type	Description
list[tuple[str, str]]	list[tuple[str,str]]: List of edges.

nodes property

Get all nodes in current UGRAPH.

Returns:

Name	Type	Description
list	list[str]	list of nodes.

num_edges property

Number of edges in current UGRAPH.

Returns:

Name	Type	Description
int	int	Number of edges

num_nodes property

Number of nodes in current UGRAPH.

Returns:

Name	Type	Description
int	int	Number of nodes

__init__(nodes=None, edges=None)

UGRAPH constructor.

Parameters:

Name	Type	Description	Default
nodes	list[str] | None	Nodes. Defaults to None.	None
edges	list[tuple[str, str]] | None	Edges. Defaults to None.	None

Source code in mixpc/graphs.py

def __init__(
    self,
    nodes: list[str] | None = None,
    edges: list[tuple[str, str]] | None = None,
) -> None:
    """UGRAPH constructor.

    Args:
        nodes (list[str] | None, optional): Nodes. Defaults to None.
        edges (list[tuple[str,str]] | None, optional): Edges. Defaults to None.
    """
    if nodes is None:
        nodes = []
    if edges is None:
        edges = []

    self._nodes: set[str] = set(nodes)
    self._edges: set[tuple[str, str]] = set()
    self._neighbors: defaultdict[str, set[str]] = defaultdict(set)

    for edge in edges:
        self._add_edge(*edge)

copy()

Return a copy of the graph.

Source code in mixpc/graphs.py

def copy(self) -> UGRAPH:
    """Return a copy of the graph."""
    return UGRAPH(nodes=list(self._nodes), edges=list(self._edges))

from_pandas_adjacency(pd_amat) classmethod

Build UGRAPH from a Pandas adjacency matrix.

Parameters:

Name	Type	Description	Default
pd_amat	DataFrame	input adjacency matrix.	required

Returns:

Type	Description
UGRAPH	UGRAPH

Source code in mixpc/graphs.py

@classmethod
def from_pandas_adjacency(cls, pd_amat: pd.DataFrame) -> UGRAPH:
    """Build UGRAPH from a Pandas adjacency matrix.

    Args:
        pd_amat (pd.DataFrame): input adjacency matrix.

    Returns:
        UGRAPH
    """
    assert pd_amat.shape[0] == pd_amat.shape[1]
    nodes = list(pd_amat.columns)

    all_connections = []
    start, end = np.where(pd_amat != 0)
    for idx, _ in enumerate(start):
        all_connections.append((pd_amat.columns[start[idx]], pd_amat.columns[end[idx]]))

    edges = [tuple(item) for item in set(frozenset(item) for item in all_connections)]

    return UGRAPH(nodes=nodes, edges=edges)

is_adjacent(i, j)

Return True if the graph contains an undirected edge between i and j.

Parameters:

Name	Type	Description	Default
i	str	node i.	required
j	str	node j.	required

Returns:

Name	Type	Description
bool	bool	True if i - j

Source code in mixpc/graphs.py

def is_adjacent(self, i: str, j: str) -> bool:
    """Return True if the graph contains an undirected edge between i and j.

    Args:
        i (str): node i.
        j (str): node j.

    Returns:
        bool: True if i - j
    """
    return (i, j) in self._edges or (j, i) in self._edges

is_clique(potential_clique)

Check every pair of nodes in potential_clique is adjacent.

Source code in mixpc/graphs.py

def is_clique(self, potential_clique: set[str]) -> bool:
    """Check every pair of nodes in potential_clique is adjacent."""
    return all(self.is_adjacent(i, j) for i, j in combinations(potential_clique, 2))

neighbors(node)

Gives all neighbors of node node.

Parameters:

Name	Type	Description	Default
node	str	node in current UGRAPH.	required

Returns:

Name	Type	Description
set	set[str]	set of neighbors.

Source code in mixpc/graphs.py

def neighbors(self, node: str) -> set[str]:
    """Gives all neighbors of node `node`.

    Args:
        node (str): node in current UGRAPH.

    Returns:
        set: set of neighbors.
    """
    if node in self._neighbors:
        return self._neighbors[node]
    else:
        return set()

remove_edge(i, j)

Removes edge in question.

Parameters:

Name	Type	Description	Default
i	str	first node	required
j	str	second node	required

Raises:

Type	Description
AssertionError	if edge does not exist

Source code in mixpc/graphs.py

def remove_edge(self, i: str, j: str) -> None:
    """Removes edge in question.

    Args:
        i (str): first node
        j (str): second node

    Raises:
        AssertionError: if edge does not exist
    """
    if not self.is_adjacent(i, j):
        raise AssertionError("Edge does not exist in current UGRAPH")

    self._edges.discard((i, j))
    self._edges.discard((j, i))
    self._neighbors[i].discard(j)
    self._neighbors[j].discard(i)

remove_node(node)

Remove a node from the graph.

Parameters:

Name	Type	Description	Default
node	str	node to remove	required

Source code in mixpc/graphs.py

def remove_node(self, node: str) -> None:
    """Remove a node from the graph.

    Args:
        node (str): node to remove
    """
    self._nodes.remove(node)

    self._edges = {(i, j) for i, j in self._edges if node not in {i, j}}

    for nbr in self._neighbors[node]:
        self._neighbors[nbr].discard(node)

    self._neighbors.pop(node, "I was never here")

show()

Plot UGRAPH.

Source code in mixpc/graphs.py

def show(self) -> None:
    """Plot UGRAPH."""
    graph = self.to_networkx()
    pos = nx.circular_layout(graph)
    nx.draw(graph, pos=pos, with_labels=True)

to_networkx()

Convert to networkx graph.

Returns:

Type	Description
Graph	nx.Graph: Undirected networkx graph.

Source code in mixpc/graphs.py

def to_networkx(self) -> nx.Graph:
    """Convert to networkx graph.

    Returns:
        nx.Graph: Undirected networkx graph.
    """
    nx_ugraph = nx.Graph()
    nx_ugraph.add_nodes_from(self.nodes)
    nx_ugraph.add_edges_from(self.edges)
    return nx_ugraph