1import numpy as np
2from causallearn.graph.GeneralGraph import GeneralGraph
3from causallearn.graph.GraphNode import GraphNode
4
5# =====================================================================
6# EDITABLE: implement run_causal_discovery below
7# =====================================================================
8def run_causal_discovery(X: np.ndarray) -> GeneralGraph:
9    """
10    Input:  X of shape (n_samples, n_variables)
11    Output: estimated CPDAG as causallearn.graph.GeneralGraph.GeneralGraph
12    """
13    nodes = [GraphNode(f"X{i + 1}") for i in range(X.shape[1])]
14    return GeneralGraph(nodes)
15# =====================================================================
... 1 more lines

1"""Evaluation harness for the causal-observational-linear-gaussian task."""
2import argparse
3import os
4import sys
5
6sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
7
8from custom_algorithm import run_causal_discovery
9from data_gen import simulate_linear_gaussian
10from metrics import compute_metrics
11
12
13def main():
14    parser = argparse.ArgumentParser(
15        description="Evaluate CPDAG recovery on synthetic linear Gaussian data."
... 49 more lines

1"""Synthetic linear Gaussian DAG data generator for CPDAG benchmarking."""
2import networkx as nx
3import numpy as np
4
5from causallearn.graph.Dag import Dag
6from causallearn.graph.GraphNode import GraphNode
7
8
9def simulate_dag(n_nodes, graph_type, seed, er_prob=0.5, sf_m=2):
10    """Return a binary adjacency matrix with convention adj[i, j] = 1 for i -> j."""
11    rng = np.random.default_rng(seed)
12    graph_seed = int(rng.integers(0, 2**31 - 1))
13
14    if graph_type == "er":
15        graph = nx.erdos_renyi_graph(n_nodes, er_prob, seed=graph_seed, directed=True)
... 57 more lines

1"""Evaluation metrics for CPDAG recovery on linear Gaussian data."""
2from causallearn.graph.AdjacencyConfusion import AdjacencyConfusion
3from causallearn.graph.ArrowConfusion import ArrowConfusion
4from causallearn.graph.Graph import Graph
5from causallearn.graph.SHD import SHD
6from causallearn.utils.DAG2CPDAG import dag2cpdag
7
8
9def _safe_div(numerator, denominator):
10    return numerator / denominator if denominator > 0 else 0.0
11
12
13def _normalize_graph_output(graph_output):
14    """Normalize algorithm output to a causallearn Graph object."""
15    if isinstance(graph_output, dict) and "G" in graph_output:
... 41 more lines