ReferenceCausal graphs

Causal graphs

Causal graphs: one mixed-graph core with typed faces (DAG, ADMG, CPDAG, PAG). Import from ergodic.graph.

Build graphs with the dag(), admg(), cpdag(), pag(), or mixed() helpers. Edges read as glyph strings, so dag(["X -> Y", "Y -> Z"]) and pag(["X o-> Y", "Z o-> Y"]) do what they look like. Separation (m_separated()), the orientation machinery (meek_closure(), fci_closure()), the projections between the faces (cpdag_of_dag(), mag_of_dag(), pag_of_mag(), mag_of_pag()), the class toolkit (possible_descendants(), visible_edges(), dags_of_cpdag(), mags_of_pag()), and the primitives the identification algorithms read (districts(), topological_order(), latent_projection()) live here too.

Also exported from ergodic: GRAPH_FORMAT, Edge, GraphKind, InvalidGraphError, Mark, MixedGraph, admg, cpdag, d_separated, dag, m_separated, mixed, pag.

GRAPH_FORMAT#

GRAPH_FORMAT: str = 'ergodic.graph/1'

The serialization format to_dict() writes and from_dict() reads.

A dict without a "format" field (written before the field existed, or by hand) is read as this version. Any other value is refused, so a reader never guesses at a layout it does not know.

Edge#

class Edge(u: str, v: str, u_mark: Mark, v_mark: Mark) -> None

One edge between two distinct nodes, with a mark at each end.

Stored canonically with u < v (string order) so an undirected pair maps to exactly one Edge. u_mark is the mark at u, v_mark at v.

Edge.make#

staticmethod Edge.make(a: str, a_mark: Mark, b: str, b_mark: Mark) -> Edge

Build an edge between a and b, normalizing endpoint order.

Edge.mark_at#

def Edge.mark_at(node: str) -> Mark

Return the mark this edge carries at node.

Edge.other#

def Edge.other(node: str) -> str

Return the endpoint that is not node.

Edge.is_directed#

property Edge.is_directed: bool

Whether the edge is ->: a tail at one end and an arrowhead at the other.

Edge.is_bidirected#

property Edge.is_bidirected: bool

Whether the edge is <->: an arrowhead at both ends.

Edge.is_undirected#

property Edge.is_undirected: bool

Whether the edge is --: a tail at both ends.

Edge.has_circle#

property Edge.has_circle: bool

Whether either end carries a circle mark (a PAG edge such as o->).

Edge.directed_pair#

def Edge.directed_pair() -> tuple[str, str] | None

Return (parent, child) if this edge is directed, else None.

Edge.glyph#

def Edge.glyph() -> str

Render the edge, e.g. "X o-> Y".

EdgeSpec#

EdgeSpec: UnionType = ergodic.graph._graph.Edge | str | tuple[typing.Any, ...]

How a caller can name an edge: an Edge, a glyph string like "X o-> Y", a 2-tuple (u, v) (read as u -> v), or a 4-tuple (u, u_mark, v, v_mark).

GraphKind#

class GraphKind(enum.Enum)

The kind of a mixed graph, which fixes the allowed marks and rules.

InvalidGraphError#

A graph violates the rules of its declared GraphKind.

Mark#

class Mark(enum.Enum)

A mark at one end of an edge.

A mixed graph edge has a mark at each end. TAIL (drawn -) and ARROW (drawn >) give directed (->), bidirected (<->), and undirected (--) edges. CIRCLE (drawn o) means the orientation is unknown, the marks an FCI search leaves on a PAG.

Mark.char#

def Mark.char(*, left: bool) -> str

Return the glyph for this mark on the left or right side of an edge.

MixedGraph#

class MixedGraph(
    nodes: frozenset[str] = <factory>,
    edges: frozenset[Edge] = <factory>,
    latent: frozenset[str] = <factory>,
    selection: frozenset[str] = <factory>,
    kind: GraphKind = <GraphKind.MIXED: 'mixed'>,
) -> None

An immutable mixed graph over named nodes.

Use the dag(), admg(), cpdag(), pag(), or mixed() builders rather than constructing this directly.

MixedGraph.node_list#

def MixedGraph.node_list() -> list[str]

Return the node names, sorted.

MixedGraph.edge_list#

def MixedGraph.edge_list() -> list[Edge]

Return the edges, sorted by their stored (u, v) endpoints.

MixedGraph.has_edge#

def MixedGraph.has_edge(a: str, b: str) -> bool

Return True if any edge joins a and b, whatever its marks.

The order of a and b does not matter.

MixedGraph.get_edge#

def MixedGraph.get_edge(a: str, b: str) -> Edge | None

Return the edge between a and b, or None when they are not adjacent.

The order of a and b does not matter. The edge keeps its stored orientation, so read its marks with mark_at().

MixedGraph.mark_at#

def MixedGraph.mark_at(a: str, b: str, node: str) -> Mark | None

Return the mark at node on the edge between a and b.

Returns: The Mark at node, or None when a and b are not adjacent.

MixedGraph.neighbors#

def MixedGraph.neighbors(x: str) -> frozenset[str]

Return every node that shares an edge with x, whatever the marks.

Raises ValueError when x is not a node.

MixedGraph.parents#

def MixedGraph.parents(x: str) -> frozenset[str]

Return the nodes with a directed edge into x (p -> x).

Raises ValueError when x is not a node.

MixedGraph.children#

def MixedGraph.children(x: str) -> frozenset[str]

Return the nodes x points to with a directed edge (x -> c).

Raises ValueError when x is not a node.

MixedGraph.spouses#

def MixedGraph.spouses(x: str) -> frozenset[str]

Return the nodes joined to x by a bidirected edge (x <-> s).

Raises ValueError when x is not a node.

MixedGraph.ancestors#

def MixedGraph.ancestors(x: str) -> frozenset[str]

Return the nodes with a directed path into x, excluding x itself.

Only directed edges count; undirected, bidirected and circle edges are not followed. Raises ValueError when x is not a node.

MixedGraph.descendants#

def MixedGraph.descendants(x: str) -> frozenset[str]

Return the nodes reachable from x along directed edges, excluding x.

Only directed edges count; undirected, bidirected and circle edges are not followed. Raises ValueError when x is not a node.

MixedGraph.markov_blanket#

def MixedGraph.markov_blanket(x: str) -> frozenset[str]

Parents, children, and the other parents of children (DAG only).

MixedGraph.add_nodes#

def MixedGraph.add_nodes(*names: str) -> MixedGraph

Return a graph with names added as isolated nodes; existing names are kept.

MixedGraph.add_edge#

def MixedGraph.add_edge(spec: EdgeSpec) -> MixedGraph

Add an edge on a free node pair; adding the edge already there is a no-op.

Raises ValueError when the pair already carries a different edge. Call remove_edge() first, or orient() to redirect.

MixedGraph.remove_edge#

def MixedGraph.remove_edge(a: str, b: str) -> MixedGraph

Return a graph without the edge between a and b.

Both nodes stay in the graph and the graph's kind is unchanged. When the two nodes are not adjacent, the graph comes back as it is.

MixedGraph.orient#

def MixedGraph.orient(parent: str, child: str) -> MixedGraph

Return a graph with the edge between the two nodes set to parent -> child.

MixedGraph.set_mark#

def MixedGraph.set_mark(
    a: str,
    b: str,
    *,
    at: str,
    mark: Mark,
) -> MixedGraph

Return a graph with the {a, b} edge's mark at endpoint at set to mark.

The finer-grained sibling of orient(): one endpoint changes, the other keeps its mark (how the FCI rules orient one mark at a time). Raises KeyError when the pair carries no edge or at is not one of its endpoints.

MixedGraph.as_kind#

def MixedGraph.as_kind(kind: GraphKind | str) -> MixedGraph

Re-tag the graph as kind (an enum or its name) after checking it fits.

MixedGraph.is_acyclic#

def MixedGraph.is_acyclic() -> bool

Return True if the directed part of the graph has no cycle.

MixedGraph.is_dag#

def MixedGraph.is_dag() -> bool

Every edge directed and no directed cycle. This check is complete.

MixedGraph.is_admg#

def MixedGraph.is_admg() -> bool

Directed or bidirected edges only, and no directed cycle. This check is complete.

MixedGraph.is_cpdag#

def MixedGraph.is_cpdag() -> bool

Return True when the graph really is a CPDAG. This check is complete.

Three things must hold: every edge is directed or undirected, some DAG has exactly this skeleton and these v-structures (the Dor-Tarsi consistent extension exists), and the graph equals the CPDAG of that extension, so every compelled edge is directed and every reversible edge is left undirected.

MixedGraph.is_mag#

def MixedGraph.is_mag() -> bool

The ancestral rules of Richardson and Spirtes (2002).

Three edge types (directed, bidirected, undirected), no directed cycle, no almost-directed cycle (a bidirected edge between ancestors), and the endpoints of any undirected edge carry no arrowhead at all (no parents and no spouses), which is how selection bias shows up in a MAG. One gap: maximality (every non-adjacent pair separable) is not checked here; the is_pag() round trip is what exercises it.

MixedGraph.is_pag#

def MixedGraph.is_pag() -> bool

Return True when the graph really is a complete PAG. This check is complete.

The round trip decides it, exactly as is_cpdag() does for CPDAGs: extract one member MAG (mag_of_pag), check it satisfies the MAG rules, and require the canonical PAG of that member (pag_of_mag) to be exactly this graph, so every invariant mark is oriented and every varying mark is a circle. Selection bias is in scope: undirected and circle-tail edges are read as Zhang's rules R5 to R7 leave them. One known gap rides along from is_mag(): maximality of the member is not checked.

MixedGraph.validate#

def MixedGraph.validate(kind: GraphKind | None = None) -> None

Check the graph against kind (default: its own), raising on failure.

MixedGraph.m_separated#

def MixedGraph.m_separated(x: Any, y: Any, given: Any = frozenset()) -> bool

Return True if x and y are m-separated given given.

Each argument is a node name or an iterable of names. On a graph with circle marks (a PAG) the check reads definite-status paths, so on a complete PAG the answer holds in every member of the class. A CPDAG-kind graph gets the same class reading, with its undirected edges read as circles.

Returns: True when no active path joins x and y given given.

MixedGraph.d_separated#

def MixedGraph.d_separated(x: Any, y: Any, given: Any = frozenset()) -> bool

Return True if x and y are d-separated given given.

The DAG case of m_separated(), with the same arguments.

MixedGraph.do#

def MixedGraph.do(targets: Any) -> MixedGraph

Graph surgery for an intervention: cut every arrowhead into targets.

MixedGraph.induced_subgraph#

def MixedGraph.induced_subgraph(nodes: Any) -> MixedGraph

Return the subgraph on nodes and the edges among them.

Names not in the graph are ignored. Latent and selection flags carry over for the nodes that stay.

MixedGraph.ancestral_subgraph#

def MixedGraph.ancestral_subgraph(nodes: Any) -> MixedGraph

Return the subgraph on nodes plus all their ancestors, with the edges among them.

Ancestry follows directed edges only. Names not in the graph are ignored.

MixedGraph.to_networkx#

def MixedGraph.to_networkx() -> Any

Return a networkx.MultiDiGraph that from_networkx() reads back.

Each edge becomes one u -> v arc with u_mark and v_mark data, and each node carries boolean latent and selection data. Needs networkx.

MixedGraph.to_adjacency#

def MixedGraph.to_adjacency(
    order: list[str] | None = None,
) -> tuple[np.ndarray, list[str]]

Return the directed adjacency matrix and its node order.

matrix[i, j] == 1 when there is a directed edge from node i to node j. Bidirected, undirected and circle edges do not appear.

Returns: A (matrix, order) pair; matrix is an integer array.

MixedGraph.to_dict#

def MixedGraph.to_dict() -> dict[str, Any]

Return a JSON-ready dict that from_dict() reads back.

The dict holds format ("ergodic.graph/1"), kind, nodes, edges (each with u, v, u_mark and v_mark), latent and selection.

MixedGraph.from_dict#

classmethod MixedGraph.from_dict(data: dict[str, Any]) -> MixedGraph

Rebuild a graph from to_dict() output and validate it.

A missing format field reads as "ergodic.graph/1". kind defaults to "mixed", and latent and selection default to empty.

Returns: The graph, checked against the rules of its kind.

MixedGraph.from_networkx#

classmethod MixedGraph.from_networkx(
    graph: Any,
    *,
    kind: GraphKind | str | None = None,
) -> MixedGraph

Build a graph from a networkx graph.

Edges with u_mark and v_mark data keep those marks; a plain edge reads as u -> v. Node latent and selection flags carry over. With kind (a GraphKind or its name) the result is re-tagged after checking it fits; without it, the result is tagged DAG when every edge is directed and acyclic, otherwise mixed.

admg#

def admg(edges: Iterable[EdgeSpec] = (), **kwargs: Any) -> MixedGraph

Build an acyclic directed mixed graph (directed and bidirected edges).

consistent_extension#

def consistent_extension(pdag: MixedGraph) -> MixedGraph | None

A DAG with exactly the PDAG's skeleton and v-structures, or None.

Dor and Tarsi (1992): repeatedly find a node x with no outgoing directed edges whose undirected neighbours are adjacent to every other neighbour of x, orient all undirected edges at x into x, and retire x. When every node retires the oriented graph is the extension; when no candidate exists none does.

cpdag#

def cpdag(edges: Iterable[EdgeSpec] = (), **kwargs: Any) -> MixedGraph

Build a CPDAG (directed and undirected edges).

cpdag_of_dag#

def cpdag_of_dag(dag: MixedGraph) -> MixedGraph

The CPDAG of a DAG: undirected skeleton, restored v-structures, Meek closure.

Every DAG in the Markov equivalence class of dag maps to the same CPDAG, which keeps an edge directed exactly when every member orients it the same way.

d_separated#

def d_separated(
    graph: MixedGraph,
    x: str | Iterable[str],
    y: str | Iterable[str],
    given: str | Iterable[str] = frozenset(),
) -> bool

d-separation, the DAG special case of m_separated().

dag#

def dag(edges: Iterable[EdgeSpec] = (), **kwargs: Any) -> MixedGraph

Build a directed acyclic graph. Edges default to directed (u -> v).

dags_of_cpdag#

def dags_of_cpdag(
    cpdag: MixedGraph,
    *,
    max_members: int | None = None,
) -> tuple[MixedGraph, ...]

Every DAG in the CPDAG's Markov equivalence class, in a deterministic order.

Branch on the first undirected edge, orient it each way, close under Meek's rules, and recurse. Distinct branches fix that edge differently, so no member appears twice and none is missed (the rules are sound, so every member agrees with the closure of its own branch). Each leaf is verified against the class definition, cpdag_of_dag(leaf) equal to the input, so a branch that closes into a non-member dies quietly and the answer is exact. max_members raises once the count would pass it; there is no silent truncation.

district_of#

def district_of(graph: MixedGraph, node: str) -> frozenset[str]

Return the district (c-component) containing node.

districts#

def districts(graph: MixedGraph) -> frozenset[frozenset[str]]

The c-components: connected components of the bidirected (<->) subgraph.

Two observed nodes share a district when a <-> path joins them; a node with no bidirected edge is its own singleton. Every node lands in exactly one district and the districts partition graph.nodes. On a DAG (no bidirected edges) every district is a singleton. Defined over the nodes as they are: project latents first (latent_projection()) if the graph still carries latent= designations.

extendable#

def extendable(pdag: MixedGraph) -> bool

Return True when the PDAG admits a consistent extension.

fci_closure#

def fci_closure(
    graph: MixedGraph,
    *,
    discriminating: Callable[[str, str, str, str], bool | None],
) -> MixedGraph

Close a partially oriented graph under Zhang's rules R1 to R10.

The input carries its v-structure arrowheads (and any knowledge-seeded marks); circles elsewhere. Rules only ever turn a circle into a tail or an arrowhead: a mark that is already definite never moves, so a seeded mark that disagrees with a rule simply stays and the conflict surfaces at the caller's validation. discriminating(d, a, b, c) answers R4's question, whether b is a collider on the discriminating path; None skips that path. Sweeps run in sorted order until a full pass changes nothing, so the closure is deterministic. R5 to R7 are the selection rules and never fire on a selection-free class; Zhang stages the rules (R1 to R4, then R5 once, then R6 to R10) but every rule is individually sound and marks only move circle to definite, so the joint fixpoint is the same closure, which the round-trip tests hold executably. Returns kind mixed.

latent_projection#

def latent_projection(
    graph: MixedGraph,
    keep: Iterable[str] | None = None,
) -> MixedGraph

Project a DAG or ADMG with latents onto the observed margin, as an ADMG.

keep defaults to graph.nodes - graph.latent. The projected-out set is H = nodes - keep. Over keep the result is an ADMG (kind=ADMG, empty latent):

• a -> b iff a directed path a -> ... -> b has every intermediate node in H.
• a <-> b iff a path between a and b has every intermediate node in H and a non-collider, with arrowheads into both a and b.

Computed by iterative single-node elimination of each node in H, which is equivalent. The defining contract is m-separation preservation: for disjoint observed A, B, Z (none in H), the projection and graph agree on m_separated(A, B, Z). This is the ADMG projection the ID algorithm needs, distinct from the MAG ancestral projection mag_of_dag().

The result can carry a bow as two parallel edges on one pair (a -> b and a <-> b), which the marks model otherwise forbids. This is correct for an ADMG, and the ID engine reads it through parents, spouses, districts, and m_separated, all of which see both edges. A projected graph is an internal ID input, not a serializable one: to_dict, to_adjacency, and to_networkx raise on a parallel-edge pair rather than drop or mangle it.

m_connected#

def m_connected(
    graph: MixedGraph,
    x: str | Iterable[str],
    y: str | Iterable[str],
    given: str | Iterable[str] = frozenset(),
) -> bool

Return True if some active path joins x and y given given.

Circle-free graphs (DAG, ADMG, MAG) use plain m-connection. A graph with circle marks (a PAG) uses definite m-connection, which on a valid complete PAG is the class-invariant answer. A CPDAG-kind graph gets the same class reading with its undirected edges recoded as circles; an undirected edge on any other kind is walked literally, as a tail at both ends.

m_separated#

def m_separated(
    graph: MixedGraph,
    x: str | Iterable[str],
    y: str | Iterable[str],
    given: str | Iterable[str] = frozenset(),
) -> bool

Return True if x and y are m-separated given given.

mag_of_dag#

def mag_of_dag(graph: MixedGraph) -> MixedGraph

Project a DAG or ADMG with latent and selection nodes onto its MAG.

Two observed nodes are adjacent exactly when no observed subset separates them given the selection nodes, decided by the one witness set that suffices (Richardson and Spirtes 2002, Theorem 4.2): the observed ancestors of either endpoint or of selection, plus the selection nodes themselves. Marks read ancestry in the full graph: a tail at X on the edge to Y means X is an ancestor of Y or of selection there, an arrowhead means it is neither; two tails make an undirected edge, which on a DAG means both endpoints are ancestors of selection. The result preserves every m-separation over the observed margin conditional on selection, which is the defining property of the projection.

mag_of_pag#

def mag_of_pag(p: MixedGraph) -> MixedGraph

Extract one member MAG from a PAG (Zhang 2008, arrowhead augmentation).

Every o-> edge becomes directed into its arrowhead, every -o edge becomes directed out of its tail (the circle takes the arrowhead), undirected and bidirected and directed edges stay as they are, and the circle-circle component is oriented into a DAG that adds no unshielded collider (the Dor-Tarsi extension of its undirected skeleton; for a valid complete PAG that component is chordal, R5 guarantees it, so the extension exists). An inextricable circle component or a result that fails the MAG rules raises. Maximality of the member is not checked here because is_mag does not check it; the is_pag round trip is what guards it.

mags_of_pag#

def mags_of_pag(
    p: MixedGraph,
    *,
    max_members: int | None = None,
    selection: bool = False,
) -> tuple[MixedGraph, ...]

Every MAG in the PAG's equivalence class, in a deterministic order.

Each circle mark takes a tail or an arrowhead; an assignment survives when it yields a valid MAG whose canonical PAG (pag_of_mag()) is exactly p. With selection=False (the default) only the members without undirected edges count: the reading every consumer that assumes no selection bias quantifies over, identification among them. Pass selection=True for the full Markov equivalence class, where a circle-circle edge may also resolve to an undirected selection edge. A PAG that itself carries selection marks (undirected or circle-tail edges) has no selection-free members and raises under the default. Exponential in the number of circle marks by design, which is fine at the scale the library targets; max_members raises once the count would pass it (no silent truncation). mag_of_pag(p) is always among the members.

meek_closure#

def meek_closure(
    pdag: MixedGraph,
    *,
    forbidden: Callable[[str, str], bool] | None = None,
    ambiguous: Collection[tuple[str, str, str]] | None = None,
) -> MixedGraph

Apply Meek's rules R1 to R4 to a PDAG until fixpoint.

Each rule turns an undirected edge into a directed one whenever the reverse orientation would create a cycle or a new v-structure in every consistent extension. forbidden(u, v) returning True blocks the orientation u -> v: the rule simply does not fire that way and the edge stays undirected unless another rule orients it.

ambiguous lists unshielded triples (x, z, y), z the middle node and the ends unordered, that the caller cannot certify as definite non-colliders (PC's conservative and majority collider rules leave such triples behind). A definite non-collider is exactly what R1 and R3 consume: R1 reads c -> a -- b with c and b nonadjacent as compelling a -> b only because the collider c -> a <- b was ruled out, and R3 rests on its two converging parents never colliding at the undirected partner. Both rules skip a gated triple; R2 (a cycle rule) and R4 run unchanged. Default None gates nothing.

The result is returned with kind mixed; re-face it with as_kind() when more is known.

mixed#

def mixed(
    edges: Iterable[EdgeSpec] = (),
    *,
    nodes: Iterable[str] = (),
    latent: Iterable[str] = (),
    selection: Iterable[str] = (),
    kind: GraphKind = <GraphKind.MIXED: 'mixed'>,
) -> MixedGraph

Build a mixed graph of the given kind and check it satisfies that kind.

pag#

def pag(edges: Iterable[EdgeSpec] = (), **kwargs: Any) -> MixedGraph

Build a partial ancestral graph, validated as a complete PAG.

The marks must be exactly what an FCI run would leave: every invariant mark oriented (closed under the orientation rules) and a circle wherever the equivalence class disagrees. Build with mixed() instead when you want circle marks without that guarantee.

pag_of_mag#

def pag_of_mag(m: MixedGraph) -> MixedGraph

The canonical PAG of a MAG: the marks every Markov-equivalent MAG shares.

Same skeleton, every mark a circle; unshielded colliders copied from the MAG (they are class invariants); then the rule closure, with R4's discriminating-path question answered by the MAG's own marks. Zhang (2008) proved the ten rules complete with latent confounders and selection bias both, so the result carries exactly the invariant marks. The input should be maximal; is_mag cannot check that, and a non-maximal input gives a PAG for the wrong class.

possible_ancestors#

def possible_ancestors(
    graph: MixedGraph,
    nodes: str | Iterable[str],
) -> frozenset[str]

Nodes with a possibly directed path into nodes.

The mirror of possible_descendants(): w counts when some member of the class can orient a path from w all the way into a seed. Each seed is excluded from its own walk.

possible_descendants#

def possible_descendants(
    graph: MixedGraph,
    nodes: str | Iterable[str],
) -> frozenset[str]

Nodes reachable from nodes along a possibly directed path.

A path is possibly directed when no edge on it carries an arrowhead at the endpoint nearer the start, so some member of the graph's class can point every edge away from the start. On a DAG or ADMG this is plain descendants. Each seed is excluded from its own walk, mirroring descendants().

topological_order#

def topological_order(graph: MixedGraph) -> tuple[str, ...]

A topological order over the directed edges, tie-broken by node name.

For every directed edge a -> b, a precedes b. Bidirected edges impose no order. The name tie-break makes the order stable across runs, so estimands that read it render reproducibly. Raises on a graph whose directed part has a cycle.

visible_edges#

def visible_edges(graph: MixedGraph) -> frozenset[tuple[str, str]]

The visible directed edges, as (tail, head) pairs (Zhang 2008).

A -> B is visible when some C not adjacent to B has an edge into A, or reaches A through a collider path into A whose every interior node is a parent of B. Visibility certifies the edge carries no hidden confounding in any represented world, which is what adjustment amenability on MAGs and PAGs consumes. The marks are read as given (arrowheads must be definite); on a DAG or CPDAG every directed edge is visible by causal sufficiency, and callers there should not need this function.