Graph types and Meek's rules¶

TimeSeriesGraph `dataclass` ¶

TimeSeriesGraph(
    n_vars: int,
    tau_max: int,
    lagged_edges: set[tuple[int, int, int]] = set(),
    changing_modules: set[int] = set(),
    var_names: list[str] | None = None,
)

Causal graph over a time-series with surrogate variable.

Edge encoding for the contemporaneous adjacency matrix contemp_adj:

contemp_adj[i, j] == 1 and contemp_adj[j, i] == 0: X_i -> X_j.
contemp_adj[i, j] == 1 and contemp_adj[j, i] == 1: X_i -- X_j (undirected; orientation not yet determined).
contemp_adj[i, j] == 0 and contemp_adj[j, i] == 0: no edge.

Lagged edges are stored as a set of (source, target, lag) tuples, always interpreted as directed X_source[t-lag] -> X_target[t].

Parameters:

Name	Type	Description	Default
`n_vars`	`int`	Number of observed variables (excluding the surrogate `C`).	required
`tau_max`	`int`	Maximum lag considered.	required

add_contemp_undirected ¶

add_contemp_undirected(i: int, j: int) -> None

Add an undirected contemporaneous edge X_i -- X_j.

remove_contemp_edge ¶

remove_contemp_edge(i: int, j: int) -> None

Remove a contemporaneous edge between X_i and X_j.

orient_contemp ¶

orient_contemp(src: int, dst: int) -> None

Orient a contemporaneous edge as X_src -> X_dst.

is_contemp_directed ¶

is_contemp_directed(i: int, j: int) -> bool

True iff there is a directed edge in either direction (not undirected).

contemp_neighbors ¶

contemp_neighbors(i: int) -> set[int]

All variables connected to i by any contemporaneous edge.

add_lagged_edge ¶

add_lagged_edge(src: int, dst: int, lag: int) -> None

Record a lagged edge X_src[t-lag] -> X_dst[t].

lagged_parents ¶

lagged_parents(j: int) -> set[tuple[int, int]]

Return {(source_var, lag)} for all lagged parents of X_j.

lagged_parents_union ¶

lagged_parents_union(vars_: list[int] | set[int]) -> set[tuple[int, int]]

Union of lagged parents across multiple targets.

mark_changing ¶

mark_changing(i: int) -> None

Mark variable i as having a changing mechanism (C -> X_i).

directed_contemp_edges ¶

directed_contemp_edges() -> Iterator[tuple[int, int]]

Yield (src, dst) for every contemporaneous edge oriented as src -> dst.

undirected_contemp_edges ¶

undirected_contemp_edges() -> Iterator[tuple[int, int]]

Yield (i, j) with i < j for each undirected contemporaneous edge.

to_networkx ¶

to_networkx() -> nx.DiGraph

Export as a NetworkX directed graph.

Nodes are tuples (var_index, "now") for current-time variables, (var_index, lag) for lagged variables (lag >= 1), and the string "C" for the surrogate node when there are changing modules. Undirected contemporaneous edges are emitted as a pair of directed edges.

summary ¶

summary() -> str

Render a human-readable summary.

apply_meek_rules ¶

apply_meek_rules(
    graph: TimeSeriesGraph, max_iter: int = 100
) -> TimeSeriesGraph

Iteratively apply Meek's rules to a contemporaneous PDAG until a fixed point.

Modifies graph in place and returns it.

Rules implemented (over contemporaneous edges only):

R1 — if a -> b and b -- c and a and c are not adjacent, orient b -> c (avoids new v-structure).
R2 — if a -> b -> c and a -- c, orient a -> c (avoids cycle).
R3 — if a -- b, a -- c, a -- d, c -> b, d -> b, and c and d are not adjacent, orient a -> b.
R4 — if a -- b, a -- c, c -> d, d -> b, and a and d are not adjacent, orient a -> b.