LON Module

LON dataclass

Local Optima Network (LON) representation.

A LON is a directed graph where nodes represent local optima and edges represent transitions between them discovered during basin-hopping search.

Attributes:

Name	Type	Description
graph	Graph	The underlying igraph Graph object. Default: empty directed ig.Graph.
best_fitness	float | None	The best fitness value found. Default: None.
final_run_values	Series | None	Series mapping run number to final fitness value. Default: None.
eq_atol	float	Tolerance for considering fitness values as equal. Default: 1e-12.
minimize	bool	Whether this is a minimization problem. Default: True.

Source code in src/lonkit/lon.py

@dataclass
class LON:
    """
    Local Optima Network (LON) representation.

    A LON is a directed graph where nodes represent local optima and edges
    represent transitions between them discovered during basin-hopping search.

    Attributes:
        graph: The underlying `igraph` Graph object. Default: empty directed `ig.Graph`.
        best_fitness: The best fitness value found. Default: `None`.
        final_run_values: `Series` mapping run number to final fitness value. Default: `None`.
        eq_atol: Tolerance for considering fitness values as equal. Default: `1e-12`.
        minimize: Whether this is a minimization problem. Default: `True`.
    """

    graph: ig.Graph = field(default_factory=lambda: ig.Graph(directed=True))
    best_fitness: float | None = None
    final_run_values: pd.Series | None = None
    eq_atol: float = DEFAULT_ATOL
    minimize: bool = True

    @classmethod
    def from_trace_data(
        cls,
        trace: pd.DataFrame,
        config: LONConfig | None = None,
    ) -> "LON":
        """
        Create a LON from trace data.

        Args:
            trace: DataFrame with one accepted transition per row, in chronological order
                within each run, and columns `[run, fit1, node1, fit2, node2]` (in any order) where:
                - run: integer run number
                - fit1: fitness value of source node
                - node1: string identifier of source node
                - fit2: fitness value of target node
                - node2: string identifier of target node
            config: Optional configuration for LON construction. If `None`, uses default
                configuration with minimum fitness aggregation. Default: `None`.

        Returns:
            `LON` instance with constructed graph.

        Note:
            Edges are kept exactly as recorded, so the LON is not required to be monotonic.
            Worsening edges can appear if the sampler accepts non-improving moves, or when
            the same node is recorded with different fitness values and these are aggregated
            (see `LONConfig.fitness_aggregation`). `CMLON.from_lon()` removes such edges.

            The trace is validated before construction. A `UserWarning` is emitted if a
            trajectory is broken, i.e. within a run `node2` of a row differs from `node1`
            of the next row, which usually means that rows are out of order or include
            rejected moves.

        Raises:
            ValueError: If the trace is malformed: it is empty, its columns differ from
                `[run, fit1, node1, fit2, node2]`, it has missing values, run numbers are
                not integers, fitness values are non-numeric or infinite, or node
                identifiers are not strings. Also raised if fitness_aggregation is `"strict"` and duplicates are
                detected, or if `max_fitness_deviation` threshold is exceeded.
        """
        config = config or LONConfig()
        trace = _validate_trace(trace)

        # Extract final fitness value from each run as a Series
        final_run_values = trace.groupby("run").tail(1).set_index("run")["fit2"]

        lnodes = pd.concat(
            [
                trace[["node1", "fit1"]].rename(columns={"node1": "Node", "fit1": "Fitness"}),
                trace[["node2", "fit2"]].rename(columns={"node2": "Node", "fit2": "Fitness"}),
            ],
            ignore_index=True,
        )

        # Node deduplication by grouping and aggregation
        node_agg = (
            lnodes.groupby("Node").agg({"Fitness": ["min", "max", "mean", "nunique"]}).reset_index()
        )
        node_agg.columns = pd.Index(
            [
                "Node",
                "Fitness_min",
                "Fitness_max",
                "Fitness_mean",
                "Fitness_nunique",
            ]
        )

        visit_counts = lnodes.groupby("Node").size().reset_index(name="Count")

        duplicates = node_agg[node_agg["Fitness_nunique"] > 1]

        # Check for duplication issues, raise errors or warning according to config
        if not duplicates.empty:
            _validate_duplicate_nodes(node_agg, duplicates, config)

        match config.fitness_aggregation:
            case "first":
                fitness_values = lnodes.groupby("Node", as_index=False).first()[["Node", "Fitness"]]
            case "max":
                fitness_values = node_agg[["Node", "Fitness_max"]].rename(
                    columns={"Fitness_max": "Fitness"}
                )
            case "mean":
                fitness_values = node_agg[["Node", "Fitness_mean"]].rename(
                    columns={"Fitness_mean": "Fitness"}
                )
            case _:  # "min" or default
                fitness_values = node_agg[["Node", "Fitness_min"]].rename(
                    columns={"Fitness_min": "Fitness"}
                )

        # Merge fitness values with visit counts
        nodes = pd.merge(fitness_values, visit_counts, on="Node")

        edges = trace.groupby(["node1", "node2"], as_index=False).size()
        edges.columns = pd.Index(["Start", "End", "Count"])

        # Remove self-loops
        edges = edges[edges["Start"] != edges["End"]]

        node_index = pd.Series(np.arange(len(nodes)), index=nodes["Node"])
        edge_list = list(
            zip(
                node_index[edges["Start"]].to_numpy().tolist(),
                node_index[edges["End"]].to_numpy().tolist(),
            )
        )

        graph = ig.Graph(n=len(nodes), edges=edge_list, directed=True)
        graph.vs["name"] = nodes["Node"].astype(str).tolist()
        graph.vs["Fitness"] = nodes["Fitness"].tolist()
        graph.vs["Count"] = nodes["Count"].tolist()
        graph.es["Count"] = edges["Count"].tolist()

        best = nodes["Fitness"].min() if config.minimize else nodes["Fitness"].max()

        return cls(
            graph=graph,
            best_fitness=best,
            final_run_values=final_run_values,
            eq_atol=config.eq_atol,
            minimize=config.minimize,
        )

    @property
    def n_vertices(self) -> int:
        """Number of vertices (local optima) in the LON."""
        return int(self.graph.vcount())

    @property
    def n_edges(self) -> int:
        """Number of edges in the LON."""
        return int(self.graph.ecount())

    @property
    def vertex_names(self) -> list[str]:
        """List of vertex names (node hashes)."""
        return list(self.graph.vs["name"]) if self.n_vertices else []

    @property
    def vertex_fitness(self) -> list[float]:
        """List of vertex fitness values."""
        return list(self.graph.vs["Fitness"]) if self.n_vertices else []

    @property
    def vertex_count(self) -> list[int]:
        """List of vertex counts (times visited)."""
        return list(self.graph.vs["Count"]) if self.n_vertices else []

    def _allclose(self, f1: float | None, f2: float | None) -> bool:
        """Check if two fitness values are equal within tolerance."""
        if f1 is None or f2 is None:
            return f1 == f2
        return np.allclose(f1, f2, atol=self.eq_atol, rtol=0.0)

    def _isclose_series(self, f1: pd.Series, f2: float):
        """Check element-wise if a `Series` of fitness values are equal to a scalar within tolerance."""
        return np.isclose(f1, f2, atol=self.eq_atol, rtol=0.0)

    def get_sinks(self) -> list[int]:
        """Get indices of sink nodes (nodes with no outgoing edges)."""
        out_degrees = self.graph.degree(mode="out")
        return [i for i, d in enumerate(out_degrees) if d == 0]

    def compute_network_metrics(self, known_best: float | None = None) -> dict[str, Any]:
        """
        Compute LON network metrics.

        Empty networks return zero for all network metrics.

        Args:
            known_best: Known global optimum value. If `None`, uses the best
                fitness found in the network. Default: `None`.

        Returns:
            Dictionary containing:
                - n_optima: Number of local optima (vertices)
                - n_edges: Number of edges
                - n_funnels: Number of funnels (sinks)
                - n_global_funnels: Number of funnels at global optimum
                - neutral: Proportion of nodes with equal-fitness connections
                - global_strength: Proportion of global optima incoming strength to total incoming strength of all nodes
                - sink_strength: Proportion of global sinks incoming strength to incoming strength of all sink nodes
        """
        best = known_best if known_best is not None else self.best_fitness

        n_optima = self.n_vertices

        sinks_id = self.get_sinks()
        n_funnels = len(sinks_id)

        sinks_fit = [self.vertex_fitness[i] for i in sinks_id]
        n_global_funnels = sum(1 for f in sinks_fit if self._allclose(f, best))

        # Neutral: proportion of nodes with equal-fitness connections
        el = self.graph.get_edgelist()
        fits = self.vertex_fitness
        neutral_edge_indices = []
        for i, (src, tgt) in enumerate(el):
            if self._allclose(fits[src], fits[tgt]):
                neutral_edge_indices.append(i)

        if neutral_edge_indices:
            gnn = self.graph.subgraph_edges(neutral_edge_indices, delete_vertices=True)
            neutral = round(gnn.vcount() / n_optima, 4)
        else:
            neutral = 0.0

        # Strength (global): incoming strength to global optima / total incoming strength
        igs = [i for i, f in enumerate(self.vertex_fitness) if self._allclose(f, best)]
        if self.n_edges > 0 and igs:
            edge_weights = self.graph.es["Count"]
            stren_igs = sum(self.graph.strength(igs, mode="in", loops=False, weights=edge_weights))
            stren_all = sum(self.graph.strength(mode="in", loops=False, weights=edge_weights))
            global_strength = round(stren_igs / stren_all, 4) if stren_all > 0 else 0.0
        else:
            global_strength = 0.0

        # Strength (sinks only): incoming strength to global sinks / incoming strength to all sinks
        global_sinks = [s for s in sinks_id if self._allclose(self.vertex_fitness[s], best)]
        local_sinks = [s for s in sinks_id if not self._allclose(self.vertex_fitness[s], best)]
        if self.n_edges > 0 and global_sinks:
            edge_weights = self.graph.es["Count"]
            sing = sum(
                self.graph.strength(global_sinks, mode="in", loops=False, weights=edge_weights)
            )
            sinl = (
                sum(self.graph.strength(local_sinks, mode="in", loops=False, weights=edge_weights))
                if local_sinks
                else 0
            )
            sink_strength = round(sing / (sing + sinl), 4) if (sing + sinl) > 0 else 0.0
        else:
            sink_strength = 0.0

        return {
            "n_optima": n_optima,
            "n_edges": self.n_edges,
            "n_funnels": n_funnels,
            "n_global_funnels": n_global_funnels,
            "neutral": neutral,
            "global_strength": global_strength,
            "sink_strength": sink_strength,
        }

    def compute_performance_metrics(self, known_best: float | None = None) -> dict[str, Any]:
        """
        Compute performance metrics based on sampling runs.

        Returns zero for both metrics when run values are absent or empty,
        or when no best fitness is available.

        Args:
            known_best: Known global optimum value. If `None`, uses the best
                fitness found in the network. Default: `None`.

        Returns:
            Dictionary containing:
                - success: Proportion of runs that reached the global optimum
                - deviation: Mean absolute deviation from the global optimum
        """
        best = known_best if known_best is not None else self.best_fitness
        if self.final_run_values is None or self.final_run_values.empty or best is None:
            return {"success": 0.0, "deviation": 0.0}

        # Success: proportion of runs that reached the global optimum
        success = self._isclose_series(self.final_run_values, best).sum() / len(
            self.final_run_values
        )

        # Deviation: mean deviation from the global optimum value
        deviation = (self.final_run_values - best).abs().mean()

        return {"success": success, "deviation": deviation}

    def compute_metrics(self, known_best: float | None = None) -> dict[str, Any]:
        """
        Compute all LON metrics (network topology + performance).

        This is a convenience method that combines both network metrics
        (topology-based) and performance metrics (run-based).

        Args:
            known_best: Known global optimum value. If `None`, uses the best
                fitness found in the network. Default: `None`.

        Returns:
            Dictionary containing all network and performance metrics:
                Network metrics: n_optima, n_edges, n_funnels, n_global_funnels, neutral, global_strength, sink_strength
                Performance metrics: success, deviation
        """
        network_metrics = self.compute_network_metrics(known_best)
        performance_metrics = self.compute_performance_metrics(known_best)
        return {**network_metrics, **performance_metrics}

    def to_cmlon(self) -> "CMLON":
        """
        Convert LON to Compressed Monotonic LON (CMLON).

        Returns:
            CMLON instance with contracted neutral nodes. An empty LON produces
            an empty CMLON, retaining the source LON and its metadata.
        """
        return CMLON.from_lon(self)

n_vertices property

n_vertices: int

Number of vertices (local optima) in the LON.

n_edges property

n_edges: int

Number of edges in the LON.

vertex_names property

vertex_names: list[str]

List of vertex names (node hashes).

vertex_fitness property

vertex_fitness: list[float]

List of vertex fitness values.

vertex_count property

vertex_count: list[int]

List of vertex counts (times visited).

from_trace_data classmethod

from_trace_data(
    trace: DataFrame, config: LONConfig | None = None
) -> LON

Create a LON from trace data.

Parameters:

Name	Type	Description	Default
trace	DataFrame	DataFrame with one accepted transition per row, in chronological order within each run, and columns [run, fit1, node1, fit2, node2] (in any order) where: - run: integer run number - fit1: fitness value of source node - node1: string identifier of source node - fit2: fitness value of target node - node2: string identifier of target node	required
config	LONConfig | None	Optional configuration for LON construction. If None, uses default configuration with minimum fitness aggregation. Default: None.	None

Returns:

Type	Description
LON	LON instance with constructed graph.

Note

Edges are kept exactly as recorded, so the LON is not required to be monotonic. Worsening edges can appear if the sampler accepts non-improving moves, or when the same node is recorded with different fitness values and these are aggregated (see LONConfig.fitness_aggregation). CMLON.from_lon() removes such edges.

The trace is validated before construction. A UserWarning is emitted if a trajectory is broken, i.e. within a run node2 of a row differs from node1 of the next row, which usually means that rows are out of order or include rejected moves.

Raises:

Type	Description
ValueError	If the trace is malformed: it is empty, its columns differ from [run, fit1, node1, fit2, node2], it has missing values, run numbers are not integers, fitness values are non-numeric or infinite, or node identifiers are not strings. Also raised if fitness_aggregation is "strict" and duplicates are detected, or if max_fitness_deviation threshold is exceeded.

Source code in src/lonkit/lon.py

@classmethod
def from_trace_data(
    cls,
    trace: pd.DataFrame,
    config: LONConfig | None = None,
) -> "LON":
    """
    Create a LON from trace data.

    Args:
        trace: DataFrame with one accepted transition per row, in chronological order
            within each run, and columns `[run, fit1, node1, fit2, node2]` (in any order) where:
            - run: integer run number
            - fit1: fitness value of source node
            - node1: string identifier of source node
            - fit2: fitness value of target node
            - node2: string identifier of target node
        config: Optional configuration for LON construction. If `None`, uses default
            configuration with minimum fitness aggregation. Default: `None`.

    Returns:
        `LON` instance with constructed graph.

    Note:
        Edges are kept exactly as recorded, so the LON is not required to be monotonic.
        Worsening edges can appear if the sampler accepts non-improving moves, or when
        the same node is recorded with different fitness values and these are aggregated
        (see `LONConfig.fitness_aggregation`). `CMLON.from_lon()` removes such edges.

        The trace is validated before construction. A `UserWarning` is emitted if a
        trajectory is broken, i.e. within a run `node2` of a row differs from `node1`
        of the next row, which usually means that rows are out of order or include
        rejected moves.

    Raises:
        ValueError: If the trace is malformed: it is empty, its columns differ from
            `[run, fit1, node1, fit2, node2]`, it has missing values, run numbers are
            not integers, fitness values are non-numeric or infinite, or node
            identifiers are not strings. Also raised if fitness_aggregation is `"strict"` and duplicates are
            detected, or if `max_fitness_deviation` threshold is exceeded.
    """
    config = config or LONConfig()
    trace = _validate_trace(trace)

    # Extract final fitness value from each run as a Series
    final_run_values = trace.groupby("run").tail(1).set_index("run")["fit2"]

    lnodes = pd.concat(
        [
            trace[["node1", "fit1"]].rename(columns={"node1": "Node", "fit1": "Fitness"}),
            trace[["node2", "fit2"]].rename(columns={"node2": "Node", "fit2": "Fitness"}),
        ],
        ignore_index=True,
    )

    # Node deduplication by grouping and aggregation
    node_agg = (
        lnodes.groupby("Node").agg({"Fitness": ["min", "max", "mean", "nunique"]}).reset_index()
    )
    node_agg.columns = pd.Index(
        [
            "Node",
            "Fitness_min",
            "Fitness_max",
            "Fitness_mean",
            "Fitness_nunique",
        ]
    )

    visit_counts = lnodes.groupby("Node").size().reset_index(name="Count")

    duplicates = node_agg[node_agg["Fitness_nunique"] > 1]

    # Check for duplication issues, raise errors or warning according to config
    if not duplicates.empty:
        _validate_duplicate_nodes(node_agg, duplicates, config)

    match config.fitness_aggregation:
        case "first":
            fitness_values = lnodes.groupby("Node", as_index=False).first()[["Node", "Fitness"]]
        case "max":
            fitness_values = node_agg[["Node", "Fitness_max"]].rename(
                columns={"Fitness_max": "Fitness"}
            )
        case "mean":
            fitness_values = node_agg[["Node", "Fitness_mean"]].rename(
                columns={"Fitness_mean": "Fitness"}
            )
        case _:  # "min" or default
            fitness_values = node_agg[["Node", "Fitness_min"]].rename(
                columns={"Fitness_min": "Fitness"}
            )

    # Merge fitness values with visit counts
    nodes = pd.merge(fitness_values, visit_counts, on="Node")

    edges = trace.groupby(["node1", "node2"], as_index=False).size()
    edges.columns = pd.Index(["Start", "End", "Count"])

    # Remove self-loops
    edges = edges[edges["Start"] != edges["End"]]

    node_index = pd.Series(np.arange(len(nodes)), index=nodes["Node"])
    edge_list = list(
        zip(
            node_index[edges["Start"]].to_numpy().tolist(),
            node_index[edges["End"]].to_numpy().tolist(),
        )
    )

    graph = ig.Graph(n=len(nodes), edges=edge_list, directed=True)
    graph.vs["name"] = nodes["Node"].astype(str).tolist()
    graph.vs["Fitness"] = nodes["Fitness"].tolist()
    graph.vs["Count"] = nodes["Count"].tolist()
    graph.es["Count"] = edges["Count"].tolist()

    best = nodes["Fitness"].min() if config.minimize else nodes["Fitness"].max()

    return cls(
        graph=graph,
        best_fitness=best,
        final_run_values=final_run_values,
        eq_atol=config.eq_atol,
        minimize=config.minimize,
    )

get_sinks

get_sinks() -> list[int]

Get indices of sink nodes (nodes with no outgoing edges).

Source code in src/lonkit/lon.py

def get_sinks(self) -> list[int]:
    """Get indices of sink nodes (nodes with no outgoing edges)."""
    out_degrees = self.graph.degree(mode="out")
    return [i for i, d in enumerate(out_degrees) if d == 0]

compute_metrics

compute_metrics(
    known_best: float | None = None,
) -> dict[str, Any]

Compute all LON metrics (network topology + performance).

This is a convenience method that combines both network metrics (topology-based) and performance metrics (run-based).

Parameters:

Name	Type	Description	Default
known_best	float | None	Known global optimum value. If None, uses the best fitness found in the network. Default: None.	None

Returns:

Type	Description
dict[str, Any]	Dictionary containing all network and performance metrics: Network metrics: n_optima, n_edges, n_funnels, n_global_funnels, neutral, global_strength, sink_strength Performance metrics: success, deviation

Source code in src/lonkit/lon.py

def compute_metrics(self, known_best: float | None = None) -> dict[str, Any]:
    """
    Compute all LON metrics (network topology + performance).

    This is a convenience method that combines both network metrics
    (topology-based) and performance metrics (run-based).

    Args:
        known_best: Known global optimum value. If `None`, uses the best
            fitness found in the network. Default: `None`.

    Returns:
        Dictionary containing all network and performance metrics:
            Network metrics: n_optima, n_edges, n_funnels, n_global_funnels, neutral, global_strength, sink_strength
            Performance metrics: success, deviation
    """
    network_metrics = self.compute_network_metrics(known_best)
    performance_metrics = self.compute_performance_metrics(known_best)
    return {**network_metrics, **performance_metrics}

to_cmlon

to_cmlon() -> CMLON

Convert LON to Compressed Monotonic LON (CMLON).

Returns:

Type	Description
CMLON	CMLON instance with contracted neutral nodes. An empty LON produces
CMLON	an empty CMLON, retaining the source LON and its metadata.

Source code in src/lonkit/lon.py

def to_cmlon(self) -> "CMLON":
    """
    Convert LON to Compressed Monotonic LON (CMLON).

    Returns:
        CMLON instance with contracted neutral nodes. An empty LON produces
        an empty CMLON, retaining the source LON and its metadata.
    """
    return CMLON.from_lon(self)

CMLON dataclass

Compressed Monotonic Local Optima Network (CMLON).

CMLON contracts nodes with equal fitness that are connected, creating a compressed representation of the fitness landscape.

Attributes:

Name	Type	Description
graph	Graph	The underlying igraph Graph object.
best_fitness	float | None	The best fitness value.
source_lon	LON | None	Reference to the original LON (optional).
eq_atol	float	Tolerance for considering fitness values as equal. Default: 1e-12.
minimize	bool	Whether this is a minimization problem. Default: True.

Source code in src/lonkit/lon.py

@dataclass
class CMLON:
    """
    Compressed Monotonic Local Optima Network (CMLON).

    CMLON contracts nodes with equal fitness that are connected,
    creating a compressed representation of the fitness landscape.

    Attributes:
        graph: The underlying igraph Graph object.
        best_fitness: The best fitness value.
        source_lon: Reference to the original LON (optional).
        eq_atol: Tolerance for considering fitness values as equal. Default: `1e-12`.
        minimize: Whether this is a minimization problem. Default: `True`.
    """

    graph: ig.Graph = field(default_factory=lambda: ig.Graph(directed=True))
    best_fitness: float | None = None
    source_lon: LON | None = None
    eq_atol: float = DEFAULT_ATOL
    minimize: bool = True

    @classmethod
    def from_lon(cls, lon: LON) -> "CMLON":
        """
        Create CMLON from LON by contracting neutral nodes.

        The compression process:
        1. Classify edges using the optimization direction and equality tolerance
        2. Remove worsening edges (with a warning)
        3. Create subgraph of equal-fitness edges
        4. Find weakly connected components
        5. Contract vertices using component membership, keeping the best fitness
           in each component (minimum for minimization, maximum for maximization)
           and the name of the vertex with that fitness (first in graph order on ties)
        6. Combine parallel edge weights
        7. Repeat steps 3-6 until no equal-fitness edges remain. Approximate equality
           is not transitive, so a contracted component can become equal to a
           neighbour it was not equal to before contraction.
        8. Remove edges that became worsening between component representatives
           (with a warning). Such edges cannot be merged, so their target may
           become an isolated sink.

        Args:
            lon: Source LON instance.

        Returns:
            CMLON with contracted neutral components.
        """
        if lon.n_edges == 0:
            cmlon_graph = lon.graph.copy()
            return cls(
                graph=cmlon_graph,
                best_fitness=lon.best_fitness,
                source_lon=lon,
                eq_atol=lon.eq_atol,
                minimize=lon.minimize,
            )

        # Create a working copy
        mlon = lon.graph.copy()
        mlon.vs["Count"] = [1] * mlon.vcount()

        f1, f2 = _edge_fitness(mlon)

        # Mark edge types
        edge_types = []
        for fit1, fit2 in zip(f1, f2):
            if lon._allclose(fit2, fit1):
                edge_types.append("equal")
            elif (fit2 < fit1) if lon.minimize else (fit2 > fit1):
                edge_types.append("improving")
            else:
                edge_types.append("worsening")
        mlon.es["type"] = edge_types

        # Remove worsening edges before compressing neutral components. They come from
        # samplers that accept worse solutions, or from merging duplicate nodes whose
        # fitness values differ slightly.
        worsening_edge_indices = [i for i, t in enumerate(edge_types) if t == "worsening"]
        if worsening_edge_indices:
            max_worsening = max(abs(f2[i] - f1[i]) for i in worsening_edge_indices)
            warnings.warn(
                f"Removed {len(worsening_edge_indices)} worsening edge(s) from the LON "
                f"(max fitness difference: {max_worsening:.6g}) to construct a monotonic CMLON. "
                "If this is unexpected, consider setting `fitness_precision`.",
                UserWarning,
                stacklevel=2,
            )
            mlon.delete_edges(worsening_edge_indices)

        # Contract neutral components until no equal-fitness edges remain. Approximate
        # equality is not transitive, so a contracted component (represented by the
        # best fitness of its vertices) can become equal to a neighbour that was not
        # equal to any of the vertices it was connected to before contraction.
        cmlon_graph = mlon
        while True:
            f_src, f_tgt = _edge_fitness(cmlon_graph)
            equal_edge_indices = np.flatnonzero(
                np.isclose(f_src, f_tgt, atol=lon.eq_atol, rtol=0.0)
            ).tolist()
            if not equal_edge_indices:
                break

            # Find weakly connected components of the equal-fitness subgraph
            gnn = cmlon_graph.subgraph_edges(equal_edge_indices, delete_vertices=False)
            nn_memb = gnn.components(mode="weak").membership

            # Contract vertices using component membership
            cmlon_graph = _contract_vertices(
                cmlon_graph,
                nn_memb,
                vertex_attr_comb={
                    "Fitness": "min" if lon.minimize else "max",
                    "Count": "sum",
                    "name": "min_fitness" if lon.minimize else "max_fitness",
                },
            )

        # Contraction can also turn an improving edge into a worsening one between
        # component representatives: approximate equality is not transitive, so
        # contraction can change an edge's classification. Such edges cannot be
        # merged and are removed.
        f_src, f_tgt = _edge_fitness(cmlon_graph)
        improving = f_tgt < f_src if lon.minimize else f_tgt > f_src
        worsening_after_compression = np.flatnonzero(~improving).tolist()
        if worsening_after_compression:
            warnings.warn(
                f"Removed {len(worsening_after_compression)} worsening edge(s) after "
                "neutral-component compression to construct a monotonic CMLON.",
                UserWarning,
                stacklevel=2,
            )
            cmlon_graph.delete_edges(worsening_after_compression)

        return cls(
            graph=cmlon_graph,
            best_fitness=lon.best_fitness,
            source_lon=lon,
            eq_atol=lon.eq_atol,
            minimize=lon.minimize,
        )

    def _allclose(self, f1: float | None, f2: float | None) -> bool:
        """Check if two fitness values are equal within tolerance."""
        if f1 is None or f2 is None:
            return f1 == f2
        return np.allclose(f1, f2, atol=self.eq_atol, rtol=0.0)

    @property
    def n_vertices(self) -> int:
        """Number of vertices in CMLON."""
        return int(self.graph.vcount())

    @property
    def n_edges(self) -> int:
        """Number of edges in CMLON."""
        return int(self.graph.ecount())

    @property
    def vertex_fitness(self) -> list[float]:
        """List of vertex fitness values."""
        return list(self.graph.vs["Fitness"]) if self.n_vertices else []

    @property
    def vertex_count(self) -> list[int]:
        """List of vertex counts (contracted nodes)."""
        return list(self.graph.vs["Count"]) if self.n_vertices else []

    def get_sinks(self) -> list[int]:
        """Get indices of sink nodes (nodes with no outgoing edges)."""
        out_degrees = self.graph.degree(mode="out")
        return [i for i, d in enumerate(out_degrees) if d == 0]

    def get_global_sinks(self) -> list[int]:
        """Get indices of global sinks (sinks at best fitness within `eq_atol`)."""
        sinks = self.get_sinks()
        fits = self.vertex_fitness
        return [s for s in sinks if self._allclose(fits[s], self.best_fitness)]

    def get_local_sinks(self) -> list[int]:
        """Get indices of local sinks (sinks not at best fitness within `eq_atol`)."""
        if self.best_fitness is None:
            return []
        global_sinks = set(self.get_global_sinks())
        return [s for s in self.get_sinks() if s not in global_sinks]

    def compute_network_metrics(self, known_best: float | None = None) -> dict[str, Any]:
        """
        Compute CMLON network metrics.

        Empty networks return zero for all network metrics.

        Args:
            known_best: Known global optimum value. If `None`, uses the best
                fitness found in the network. Default: `None`.

        Returns:
            Dictionary containing:
                - n_optima: Number of optima in CMLON
                - n_edges: Number of edges in CMLON
                - n_funnels: Number of funnels (sinks)
                - n_global_funnels: Number of funnels at global optimum
                - neutral: Proportion of contracted nodes
                - global_strength: Proportion of global sinks incoming strength to total incoming strength of all nodes
                - sink_strength: Proportion of global sinks incoming strength to incoming strength of all sink nodes
                - global_funnel_proportion: Proportion of nodes that can reach
                  a global optimum
        """
        best = known_best if known_best is not None else self.best_fitness

        n_optima = self.n_vertices

        sinks_id = self.get_sinks()
        n_funnels = len(sinks_id)

        sinks_fit = [self.vertex_fitness[i] for i in sinks_id]
        n_global_funnels = sum(1 for f in sinks_fit if self._allclose(f, best))

        # Neutral: proportion of contracted nodes
        if self.source_lon is not None and self.source_lon.n_vertices > 0:
            neutral = round(1.0 - self.n_vertices / self.source_lon.n_vertices, 4)
        else:
            neutral = 0.0

        # Strength (global): incoming strength to global sinks / total incoming strength
        igs = [s for s, f in zip(sinks_id, sinks_fit) if self._allclose(f, best)]
        ils = [s for s, f in zip(sinks_id, sinks_fit) if not self._allclose(f, best)]

        if self.n_edges > 0:
            edge_weights = self.graph.es["Count"]
            sing = (
                sum(self.graph.strength(igs, mode="in", loops=False, weights=edge_weights))
                if igs
                else 0
            )
            total = sum(self.graph.strength(mode="in", loops=False, weights=edge_weights))
            global_strength = round(sing / total, 4) if total > 0 else 0.0
        else:
            global_strength = 0.0

        # Strength (sinks only): incoming strength to global sinks / incoming strength to all sinks
        if self.n_edges > 0 and igs:
            edge_weights = self.graph.es["Count"]
            sinl = (
                sum(self.graph.strength(ils, mode="in", loops=False, weights=edge_weights))
                if ils
                else 0
            )
            sink_strength = round(sing / (sing + sinl), 4) if (sing + sinl) > 0 else 0.0
        else:
            sink_strength = 0.0

        gfunnel = self._compute_global_funnel_proportion(igs)

        return {
            "n_optima": n_optima,
            "n_edges": self.n_edges,
            "n_funnels": n_funnels,
            "n_global_funnels": n_global_funnels,
            "neutral": neutral,
            "global_strength": global_strength,
            "sink_strength": sink_strength,
            "global_funnel_proportion": gfunnel,
        }

    def _compute_global_funnel_proportion(self, igs: list[int]) -> float:
        """Compute proportion of nodes that can reach any of the supplied global sinks."""
        if not igs:
            return 0.0

        # Get all nodes that can reach any global sink
        reachable = set()
        for sink in igs:
            component = self.graph.subcomponent(sink, mode="in")
            reachable.update(component)

        return len(reachable) / self.n_vertices if self.n_vertices > 0 else 0.0

    def compute_performance_metrics(self, known_best: float | None = None) -> dict[str, Any]:
        """
        Compute performance metrics from the source LON.

        CMLON delegates to its source LON for performance metrics since
        it doesn't have its own sampling run data.

        Args:
            known_best: Known global optimum value. If `None`, uses the best
                fitness found in the network. Default: `None`.

        Returns:
            Dictionary containing performance metrics from source LON, or
            empty dict if no source LON is available.
        """
        return (
            self.source_lon.compute_performance_metrics(known_best)
            if self.source_lon is not None
            else {}
        )

    def compute_metrics(self, known_best: float | None = None) -> dict[str, Any]:
        """
        Compute all CMLON metrics (network topology + performance).

        This is a convenience method that combines both CMLON-specific network
        metrics and performance metrics from the source LON.

        Args:
            known_best: Known global optimum value. If `None`, uses the best
                fitness found in the network. Default: `None`.

        Returns:
            Dictionary containing all network and performance metrics:
                Network metrics: n_optima, n_edges, n_funnels, n_global_funnels, neutral,
                    global_strength, sink_strength, global_funnel_proportion
                Performance metrics: success, deviation (from source LON)
        """
        network_metrics = self.compute_network_metrics(known_best)
        performance_metrics = self.compute_performance_metrics(known_best)
        return {**network_metrics, **performance_metrics}

n_vertices property

n_vertices: int

Number of vertices in CMLON.

n_edges property

n_edges: int

Number of edges in CMLON.

vertex_fitness property

vertex_fitness: list[float]

List of vertex fitness values.

vertex_count property

vertex_count: list[int]

List of vertex counts (contracted nodes).

from_lon classmethod

from_lon(lon: LON) -> CMLON

Create CMLON from LON by contracting neutral nodes.

The compression process: 1. Classify edges using the optimization direction and equality tolerance 2. Remove worsening edges (with a warning) 3. Create subgraph of equal-fitness edges 4. Find weakly connected components 5. Contract vertices using component membership, keeping the best fitness in each component (minimum for minimization, maximum for maximization) and the name of the vertex with that fitness (first in graph order on ties) 6. Combine parallel edge weights 7. Repeat steps 3-6 until no equal-fitness edges remain. Approximate equality is not transitive, so a contracted component can become equal to a neighbour it was not equal to before contraction. 8. Remove edges that became worsening between component representatives (with a warning). Such edges cannot be merged, so their target may become an isolated sink.

Parameters:

Name	Type	Description	Default
lon	LON	Source LON instance.	required

Returns:

Type	Description
CMLON	CMLON with contracted neutral components.

Source code in src/lonkit/lon.py

@classmethod
def from_lon(cls, lon: LON) -> "CMLON":
    """
    Create CMLON from LON by contracting neutral nodes.

    The compression process:
    1. Classify edges using the optimization direction and equality tolerance
    2. Remove worsening edges (with a warning)
    3. Create subgraph of equal-fitness edges
    4. Find weakly connected components
    5. Contract vertices using component membership, keeping the best fitness
       in each component (minimum for minimization, maximum for maximization)
       and the name of the vertex with that fitness (first in graph order on ties)
    6. Combine parallel edge weights
    7. Repeat steps 3-6 until no equal-fitness edges remain. Approximate equality
       is not transitive, so a contracted component can become equal to a
       neighbour it was not equal to before contraction.
    8. Remove edges that became worsening between component representatives
       (with a warning). Such edges cannot be merged, so their target may
       become an isolated sink.

    Args:
        lon: Source LON instance.

    Returns:
        CMLON with contracted neutral components.
    """
    if lon.n_edges == 0:
        cmlon_graph = lon.graph.copy()
        return cls(
            graph=cmlon_graph,
            best_fitness=lon.best_fitness,
            source_lon=lon,
            eq_atol=lon.eq_atol,
            minimize=lon.minimize,
        )

    # Create a working copy
    mlon = lon.graph.copy()
    mlon.vs["Count"] = [1] * mlon.vcount()

    f1, f2 = _edge_fitness(mlon)

    # Mark edge types
    edge_types = []
    for fit1, fit2 in zip(f1, f2):
        if lon._allclose(fit2, fit1):
            edge_types.append("equal")
        elif (fit2 < fit1) if lon.minimize else (fit2 > fit1):
            edge_types.append("improving")
        else:
            edge_types.append("worsening")
    mlon.es["type"] = edge_types

    # Remove worsening edges before compressing neutral components. They come from
    # samplers that accept worse solutions, or from merging duplicate nodes whose
    # fitness values differ slightly.
    worsening_edge_indices = [i for i, t in enumerate(edge_types) if t == "worsening"]
    if worsening_edge_indices:
        max_worsening = max(abs(f2[i] - f1[i]) for i in worsening_edge_indices)
        warnings.warn(
            f"Removed {len(worsening_edge_indices)} worsening edge(s) from the LON "
            f"(max fitness difference: {max_worsening:.6g}) to construct a monotonic CMLON. "
            "If this is unexpected, consider setting `fitness_precision`.",
            UserWarning,
            stacklevel=2,
        )
        mlon.delete_edges(worsening_edge_indices)

    # Contract neutral components until no equal-fitness edges remain. Approximate
    # equality is not transitive, so a contracted component (represented by the
    # best fitness of its vertices) can become equal to a neighbour that was not
    # equal to any of the vertices it was connected to before contraction.
    cmlon_graph = mlon
    while True:
        f_src, f_tgt = _edge_fitness(cmlon_graph)
        equal_edge_indices = np.flatnonzero(
            np.isclose(f_src, f_tgt, atol=lon.eq_atol, rtol=0.0)
        ).tolist()
        if not equal_edge_indices:
            break

        # Find weakly connected components of the equal-fitness subgraph
        gnn = cmlon_graph.subgraph_edges(equal_edge_indices, delete_vertices=False)
        nn_memb = gnn.components(mode="weak").membership

        # Contract vertices using component membership
        cmlon_graph = _contract_vertices(
            cmlon_graph,
            nn_memb,
            vertex_attr_comb={
                "Fitness": "min" if lon.minimize else "max",
                "Count": "sum",
                "name": "min_fitness" if lon.minimize else "max_fitness",
            },
        )

    # Contraction can also turn an improving edge into a worsening one between
    # component representatives: approximate equality is not transitive, so
    # contraction can change an edge's classification. Such edges cannot be
    # merged and are removed.
    f_src, f_tgt = _edge_fitness(cmlon_graph)
    improving = f_tgt < f_src if lon.minimize else f_tgt > f_src
    worsening_after_compression = np.flatnonzero(~improving).tolist()
    if worsening_after_compression:
        warnings.warn(
            f"Removed {len(worsening_after_compression)} worsening edge(s) after "
            "neutral-component compression to construct a monotonic CMLON.",
            UserWarning,
            stacklevel=2,
        )
        cmlon_graph.delete_edges(worsening_after_compression)

    return cls(
        graph=cmlon_graph,
        best_fitness=lon.best_fitness,
        source_lon=lon,
        eq_atol=lon.eq_atol,
        minimize=lon.minimize,
    )

get_sinks

get_sinks() -> list[int]

Get indices of sink nodes (nodes with no outgoing edges).

Source code in src/lonkit/lon.py

def get_sinks(self) -> list[int]:
    """Get indices of sink nodes (nodes with no outgoing edges)."""
    out_degrees = self.graph.degree(mode="out")
    return [i for i, d in enumerate(out_degrees) if d == 0]

get_global_sinks

get_global_sinks() -> list[int]

Get indices of global sinks (sinks at best fitness within eq_atol).

Source code in src/lonkit/lon.py

def get_global_sinks(self) -> list[int]:
    """Get indices of global sinks (sinks at best fitness within `eq_atol`)."""
    sinks = self.get_sinks()
    fits = self.vertex_fitness
    return [s for s in sinks if self._allclose(fits[s], self.best_fitness)]

get_local_sinks

get_local_sinks() -> list[int]

Get indices of local sinks (sinks not at best fitness within eq_atol).

Source code in src/lonkit/lon.py

def get_local_sinks(self) -> list[int]:
    """Get indices of local sinks (sinks not at best fitness within `eq_atol`)."""
    if self.best_fitness is None:
        return []
    global_sinks = set(self.get_global_sinks())
    return [s for s in self.get_sinks() if s not in global_sinks]

compute_metrics

compute_metrics(
    known_best: float | None = None,
) -> dict[str, Any]

Compute all CMLON metrics (network topology + performance).

This is a convenience method that combines both CMLON-specific network metrics and performance metrics from the source LON.

Parameters:

Name	Type	Description	Default
known_best	float | None	Known global optimum value. If None, uses the best fitness found in the network. Default: None.	None

Returns:

Type	Description
dict[str, Any]	Dictionary containing all network and performance metrics: Network metrics: n_optima, n_edges, n_funnels, n_global_funnels, neutral, global_strength, sink_strength, global_funnel_proportion Performance metrics: success, deviation (from source LON)

Source code in src/lonkit/lon.py

def compute_metrics(self, known_best: float | None = None) -> dict[str, Any]:
    """
    Compute all CMLON metrics (network topology + performance).

    This is a convenience method that combines both CMLON-specific network
    metrics and performance metrics from the source LON.

    Args:
        known_best: Known global optimum value. If `None`, uses the best
            fitness found in the network. Default: `None`.

    Returns:
        Dictionary containing all network and performance metrics:
            Network metrics: n_optima, n_edges, n_funnels, n_global_funnels, neutral,
                global_strength, sink_strength, global_funnel_proportion
            Performance metrics: success, deviation (from source LON)
    """
    network_metrics = self.compute_network_metrics(known_best)
    performance_metrics = self.compute_performance_metrics(known_best)
    return {**network_metrics, **performance_metrics}

LONConfig dataclass

Configuration for LON construction from trace data.

Attributes:

Name	Type	Description
fitness_aggregation	Literal['min', 'max', 'mean', 'first', 'strict']	Strategy for handling nodes with multiple fitness values: - "min": Use minimum fitness - "max": Use maximum fitness - "mean": Use average fitness - "first": Use first occurrence - "strict": Raise error if duplicates are detected Default: "min".
warn_on_duplicates	bool	Whether to emit a warning when duplicate nodes detected. Default: True
max_fitness_deviation	float | None	If set, raise error if fitness deviation exceeds this threshold. Default: None (no threshold). Useful for detecting data quality issues.
eq_atol	float	Tolerance for considering fitness values as equal. Default: 1e-12.
minimize	bool	Whether this is a minimization problem. Default: True.

Source code in src/lonkit/lon.py

@dataclass
class LONConfig:
    """
    Configuration for LON construction from trace data.

    Attributes:
        fitness_aggregation: Strategy for handling nodes with multiple fitness values:
            - `"min"`: Use minimum fitness
            - `"max"`: Use maximum fitness
            - `"mean"`: Use average fitness
            - `"first"`: Use first occurrence
            - `"strict"`: Raise error if duplicates are detected
            Default: `"min"`.
        warn_on_duplicates: Whether to emit a warning when duplicate nodes detected. Default: `True`
        max_fitness_deviation: If set, raise error if fitness deviation exceeds this threshold. Default: `None` (no threshold).
            Useful for detecting data quality issues.
        eq_atol: Tolerance for considering fitness values as equal. Default: `1e-12`.
        minimize: Whether this is a minimization problem. Default: `True`.
    """

    fitness_aggregation: Literal["min", "max", "mean", "first", "strict"] = "min"
    warn_on_duplicates: bool = True
    max_fitness_deviation: float | None = None
    eq_atol: float = DEFAULT_ATOL
    minimize: bool = True

fitness_aggregation class-attribute instance-attribute

fitness_aggregation: Literal[
    "min", "max", "mean", "first", "strict"
] = "min"

warn_on_duplicates class-attribute instance-attribute

warn_on_duplicates: bool = True

max_fitness_deviation class-attribute instance-attribute

max_fitness_deviation: float | None = None

eq_atol class-attribute instance-attribute

eq_atol: float = DEFAULT_ATOL

minimize class-attribute instance-attribute

minimize: bool = True