spatial_graph_algorithms.denoise

Score, filter, and evaluate likely false edges in SpatialGraph objects.

Usage

from spatial_graph_algorithms.denoise import EdgeFilterer, denoise, score_edges
from spatial_graph_algorithms.metrics import evaluate_denoising

scores = score_edges(sg, method="jaccard")
sg_clean = EdgeFilterer().by_fraction(
    sg, scores, fraction=0.10, method="jaccard"
)
report = evaluate_denoising(sg_clean, scores=scores, method="jaccard")

For one-call filtering, pass a method name directly to denoise:

sg_clean = denoise(
    sg,
    method="local_walk_support",
    fraction_to_remove=0.05,
    sample_fraction=0.25,
    walk_len=8,
    walks_per_edge=500,
)

method="local_walk_support" uses the ES.py random-walk support scorer. Lower scores are more suspicious. For sampling diagnostics and threshold summaries, import the full result helper:

from spatial_graph_algorithms.denoise.ES import local_walk_support_result

result = local_walk_support_result(sg.adjacency_matrix, sample_fraction=0.10)
summary = result.threshold_summary(threshold=1.0)

Available scoring methods are betweenness, jaccard, square_bipartite, community, walk_visit_expansion, and local_walk_support. Polarity is stored in SCORE_POLARITY, and short method descriptions are stored in SCORING_METHODS. The old names random_walk and edge_support remain as compatibility aliases.

API Reference

spatial_graph_algorithms.denoise.SCORE_POLARITY = {'betweenness': 'positive', 'jaccard': 'negative', 'square_bipartite': 'negative', 'community': 'negative', 'walk_visit_expansion': 'negative', 'local_walk_support': 'negative', 'random_walk': 'negative', 'edge_support': 'negative'} module-attribute

spatial_graph_algorithms.denoise.SCORING_METHODS = {'betweenness': ScoringMethodInfo('positive', 'Edge betweenness centrality; high bridge centrality is suspicious.'), 'jaccard': ScoringMethodInfo('negative', 'Neighbourhood-overlap score for unipartite graphs; low overlap is suspicious.'), 'square_bipartite': ScoringMethodInfo('negative', 'Bipartite 4-cycle support; low square count is suspicious.'), 'community': ScoringMethodInfo('negative', 'Louvain intra-community consistency over repeated runs; low consistency is suspicious.'), 'walk_visit_expansion': ScoringMethodInfo('negative', 'Bridge-filtered walk visitation matrix with repeated expansion; low visitation is suspicious.'), 'local_walk_support': ScoringMethodInfo('negative', 'Candidate-edge endpoint support from local random walks; low support ratio is suspicious.')} module-attribute

spatial_graph_algorithms.denoise.EdgeScorer

Compute edge suspicion scores from graph topology.

Each method accepts a CSR adjacency matrix and returns a dict mapping canonical edge tuples (min(u, v), max(u, v)) to float scores. Score polarity (high = suspect vs low = suspect) is method-specific — see :data:SCORE_POLARITY and each method's docstring.

Methods:

Name	Description
betweenness	Edge betweenness centrality. High = suspect.
jaccard	Jaccard neighbourhood overlap. Low = suspect. Unipartite only.
square_bipartite	4-cycle count. Low = suspect. Bipartite only.
community	Louvain intra-community consistency. Low = suspect.
walk_visit_expansion	Bridge-filtered walk visitation expansion. Low = suspect.
local_walk_support	Local endpoint random-walk support ratio. Low = suspect.

Source code in src/spatial_graph_algorithms/denoise/__init__.py

class EdgeScorer:
    """Compute edge suspicion scores from graph topology.

    Each method accepts a CSR adjacency matrix and returns a dict mapping
    canonical edge tuples ``(min(u, v), max(u, v))`` to float scores.
    Score polarity (high = suspect vs low = suspect) is method-specific —
    see :data:`SCORE_POLARITY` and each method's docstring.

    Methods
    -------
    betweenness(adj)
        Edge betweenness centrality. High = suspect.
    jaccard(adj)
        Jaccard neighbourhood overlap. Low = suspect. Unipartite only.
    square_bipartite(adj)
        4-cycle count. Low = suspect. Bipartite only.
    community(adj, *, n_runs)
        Louvain intra-community consistency. Low = suspect.
    walk_visit_expansion(adj, *, walks_per_node, return_chance, seed)
        Bridge-filtered walk visitation expansion. Low = suspect.
    local_walk_support(adj, *, walk_len, walks_per_edge, sample_fraction)
        Local endpoint random-walk support ratio. Low = suspect.
    """

    def betweenness(
        self,
        adj: scipy.sparse.csr_matrix,
    ) -> dict[tuple[int, int], float]:
        """Score edges by edge betweenness centrality.

        High centrality signals a critical bridge — a characteristic of
        long-range false connections that span otherwise distinct communities.

        Parameters
        ----------
        adj : scipy.sparse.csr_matrix
            Symmetric unweighted adjacency matrix.

        Returns
        -------
        dict[tuple[int, int], float]
            Edge → normalised betweenness in ``[0, 1]``.  High = suspect.

        Notes
        -----
        Uses ``networkx.edge_betweenness_centrality``, runtime ``O(VE)``.
        Slow on graphs with more than a few thousand nodes.

        Examples
        --------
        >>> from spatial_graph_algorithms.simulate import generate
        >>> from spatial_graph_algorithms.denoise import EdgeScorer
        >>> sg = generate(n=100, seed=0)
        >>> scores = EdgeScorer().betweenness(sg.adjacency_matrix)
        >>> all(0.0 <= v <= 1.0 for v in scores.values())
        True
        """
        G = nx.from_scipy_sparse_array(adj)
        centrality = nx.edge_betweenness_centrality(G, normalized=True)
        return {_canonical(u, v): s for (u, v), s in centrality.items()}

    def jaccard(
        self,
        adj: scipy.sparse.csr_matrix,
    ) -> dict[tuple[int, int], float]:
        """Score edges by Jaccard neighbourhood overlap.

        Edges connecting nodes with few shared neighbours are more likely to
        be false connections bridging unrelated graph regions.

        Parameters
        ----------
        adj : scipy.sparse.csr_matrix
            Symmetric unweighted adjacency matrix for a **unipartite** graph.

        Returns
        -------
        dict[tuple[int, int], float]
            Edge → Jaccard score in ``[0, 1]``.  Low = suspect.

        Raises
        ------
        ValueError
            If *adj* represents a bipartite graph.

        Examples
        --------
        >>> from spatial_graph_algorithms.simulate import generate
        >>> from spatial_graph_algorithms.denoise import EdgeScorer
        >>> sg = generate(n=100, seed=0)
        >>> scores = EdgeScorer().jaccard(sg.adjacency_matrix)
        >>> all(0.0 <= v <= 1.0 for v in scores.values())
        True
        """
        G = nx.from_scipy_sparse_array(adj)
        if nx.is_bipartite(G):
            raise ValueError(
                "jaccard scorer is designed for unipartite graphs. "
                "Use square_bipartite for bipartite graphs."
            )
        scores: dict[tuple[int, int], float] = {}
        rows, cols = adj.nonzero()
        for r, c in zip(rows, cols):
            if r >= c:
                continue
            nb_u = set(adj[r].indices)
            nb_v = set(adj[c].indices)
            union = len(nb_u | nb_v)
            scores[(int(r), int(c))] = len(nb_u & nb_v) / union if union > 0 else 0.0
        return scores

    def square_bipartite(
        self,
        adj: scipy.sparse.csr_matrix,
    ) -> dict[tuple[int, int], float]:
        """Score edges by 4-cycle participation in a bipartite graph.

        True edges in a spatial bipartite graph typically close many 4-cycles
        (shared neighbours in the opposite partition).  False edges span
        unrelated partitions and form fewer squares.

        Parameters
        ----------
        adj : scipy.sparse.csr_matrix
            Symmetric unweighted adjacency matrix for a **bipartite** graph.

        Returns
        -------
        dict[tuple[int, int], float]
            Edge → 4-cycle count (as float).  Low = suspect.

        Raises
        ------
        ValueError
            If *adj* does not represent a bipartite graph.

        Examples
        --------
        >>> from spatial_graph_algorithms.simulate import generate
        >>> from spatial_graph_algorithms.denoise import EdgeScorer
        >>> sg = generate(n=80, mode="knn_bipartite", seed=0)
        >>> scores = EdgeScorer().square_bipartite(sg.adjacency_matrix)
        >>> all(v >= 0 for v in scores.values())
        True
        """
        G = nx.from_scipy_sparse_array(adj)
        if not nx.is_bipartite(G):
            raise ValueError(
                "square_bipartite scorer requires a bipartite graph. "
                "Use jaccard for unipartite graphs."
            )
        top_nodes, bottom_nodes = nx.bipartite.sets(G)
        top_adj = {u: set(G[u]) for u in top_nodes}
        bottom_adj = {v: set(G[v]) for v in bottom_nodes}

        edge_scores: dict[tuple[int, int], float] = {}
        for u, v in G.edges():
            if u in bottom_nodes:
                u, v = v, u  # ensure u ∈ top, v ∈ bottom
            b_u = top_adj[u] - {v}
            t_v = bottom_adj[v] - {u}
            count = sum(len(top_adj[w].intersection(b_u)) for w in t_v)
            edge_scores[_canonical(u, v)] = float(count)
        return edge_scores

    def square_bipartite_linalg(
        self,
        adj: scipy.sparse.csr_matrix,
    ) -> dict[tuple[int, int], float]:
        """Alternative square_bipartite implementation using sparse matrix multiplication."""
        # W3 = A @ A @ A - diag(A @ A) @ A - A @ diag(A @ A) + A.multiply(A).multiply(A)
        # (total 3-paths) - (i->k->i->j) - (i->j->k->j) + (i->j->i->j)
        adj      = adj.astype(np.int64)

        adj2     = adj @ adj
        adj2_diag   = adj2.diagonal().astype(np.int64)

        adj3     = adj2 @ adj

        w3 = (
            adj3
            - scipy.sparse.diags(adj2_diag, format="csr", dtype=np.int64) @ adj
            - adj @ scipy.sparse.diags(adj2_diag, format="csr", dtype=np.int64)
            + adj.multiply(adj).multiply(adj)
        )

        # only take scores if there is an edge in the original graph
        w3 = w3.multiply(adj > 0)

        # align w3 values with adj edges
        scores: dict[tuple[int, int], float] = {}
        rows, cols = adj.nonzero()
        for r, c in zip(rows, cols):
            scores[(int(r), int(c))] = float(w3[r, c])
        return scores


    def community(
        self,
        adj: scipy.sparse.csr_matrix,
        *,
        n_runs: int = 5,
    ) -> dict[tuple[int, int], float]:
        """Score edges by Louvain community membership consistency.

        Runs the Louvain algorithm (via igraph) *n_runs* times.  Each edge
        is scored by the fraction of runs it was classified as intra-community.
        Edges that persistently bridge communities are likely false connections.

        Parameters
        ----------
        adj : scipy.sparse.csr_matrix
            Symmetric unweighted adjacency matrix.
        n_runs : int
            Independent Louvain repetitions.  Higher = more stable scores.
            Default 5.

        Returns
        -------
        dict[tuple[int, int], float]
            Edge → intra-community fraction in ``[0, 1]``.  Low = suspect.

        Examples
        --------
        >>> from spatial_graph_algorithms.simulate import generate
        >>> from spatial_graph_algorithms.denoise import EdgeScorer
        >>> sg = generate(n=100, seed=0)
        >>> scores = EdgeScorer().community(sg.adjacency_matrix, n_runs=3)
        >>> all(0.0 <= v <= 1.0 for v in scores.values())
        True
        """
        rows, cols = adj.nonzero()
        ig_edges = [(int(r), int(c)) for r, c in zip(rows, cols) if r < c]
        G_ig = _igraph.Graph(n=adj.shape[0], edges=ig_edges, directed=False)

        canonical_edges = {
            (int(min(r, c)), int(max(r, c)))
            for r, c in zip(rows, cols)
            if r != c
        }
        within: dict[tuple[int, int], int] = {e: 0 for e in canonical_edges}

        for _ in range(n_runs):
            partition = G_ig.community_multilevel()
            membership = partition.membership
            for e in canonical_edges:
                if membership[e[0]] == membership[e[1]]:
                    within[e] += 1

        return {e: cnt / n_runs for e, cnt in within.items()}

    def walk_visit_expansion(
        self,
        adj: scipy.sparse.csr_matrix,
        *,
        walks_per_node: int = 32,
        return_chance: float = 0.3,
        expansions: int = 1,
        seed: int | None = None,
    ) -> dict[tuple[int, int], float]:
        """Score edges by bridge-filtered walk visitation expansion.

        Performs *walks_per_node* random walks from every node.  At each step
        there is a *return_chance* of attempting to return home.  A walk only
        counts when the current position has **two edge-disjoint paths** back
        to the origin in the locally discovered subgraph — edges that act as
        bridges (typical of false connections) will have their walks discarded
        more often, yielding lower scores.

        After normalisation, the visitation matrix is raised to the power
        ``2 ** expansions`` via repeated matrix squaring (*expansions* = 1
        computes ``Y @ Y``), amplifying score separation between true and
        false edges.

        Parameters
        ----------
        adj : scipy.sparse.csr_matrix
            Symmetric unweighted adjacency matrix.
        walks_per_node : int
            Walks launched per origin node.  Default 32.
        return_chance : float
            Per-step probability of attempting a return.  Default 0.3.
        expansions : int
            Number of matrix-squaring steps applied after normalisation.
            Default 1 (computes ``Y @ Y``).  Set to 0 to skip.
        seed : int, optional
            Random seed for reproducibility.

        Returns
        -------
        dict[tuple[int, int], float]
            Edge → normalised visitation frequency.  Low = suspect.

        Warns
        -----
        ResourceWarning
            When *adj* has more than 2 000 nodes (O(n²) memory footprint).

        Examples
        --------
        >>> from spatial_graph_algorithms.simulate import generate
        >>> from spatial_graph_algorithms.denoise import EdgeScorer
        >>> sg = generate(n=60, seed=0)
        >>> scores = EdgeScorer().walk_visit_expansion(
        ...     sg.adjacency_matrix, walks_per_node=4, seed=0
        ... )
        >>> len(scores) == sg.n_edges
        True
        """
        import warnings

        n = adj.shape[0]
        if n > 2000:
            warnings.warn(
                f"walk_visit_expansion scorer allocates a ({n}×{n}) visitation matrix "
                f"({n * n * 8 / 1e6:.0f} MB). "
                "Consider betweenness or jaccard for large graphs.",
                ResourceWarning,
                stacklevel=2,
            )

        rng = np.random.default_rng(seed)
        G = nx.from_scipy_sparse_array(adj)
        adj_list = [list(G.neighbors(i)) for i in range(n)]
        visit = np.zeros((n, n), dtype=np.float64)

        for origin in range(n):
            for _ in range(walks_per_node):
                current = origin
                path: list[tuple[int, int]] = []
                discovered = nx.Graph()
                discovered.add_node(origin)

                while True:
                    nbs = adj_list[current]
                    if not nbs:
                        break
                    nxt = int(rng.choice(nbs))
                    path.append((current, nxt))

                    discovered.add_edge(current, nxt, weight=1)

                    # 1-hop BFS to enrich the discovered subgraph (matches
                    # the original's mini-BFS with range(1))
                    for nb in adj_list[nxt]:
                        if not discovered.has_edge(nxt, nb):
                            discovered.add_edge(nxt, nb, weight=1)

                    if rng.random() < return_chance:
                        # A walk only counts when two edge-disjoint paths exist
                        # from the current position back to the origin.
                        # Find the first path, remove its edges, then verify a
                        # second path still exists — this selectively discards
                        # walks that traversed bridges or near-bridges.
                        temp = discovered.copy()
                        try:
                            first = nx.shortest_path(temp, source=current,
                                                     target=origin,
                                                     weight="weight")
                            for k in range(len(first) - 1):
                                temp.remove_edge(first[k], first[k + 1])
                            nx.shortest_path(temp, source=current,
                                             target=origin, weight="weight")
                        except nx.NetworkXNoPath:
                            path = []
                        break

                    current = nxt

                for src, dst in path:
                    visit[origin, dst] += 1
                    visit[origin, src] += 1

        # Row-normalise
        row_sums = visit.sum(axis=1)
        with np.errstate(divide="ignore", invalid="ignore"):
            norm_factors = np.where(row_sums > 0, 1.0 / row_sums, 0.0)
        Y = visit * norm_factors[:, np.newaxis]

        # Matrix-squaring expansion: amplifies score separation
        for _ in range(expansions):
            Y = np.dot(Y, Y)

        scores: dict[tuple[int, int], float] = {}
        rows, cols = adj.nonzero()
        for r, c in zip(rows, cols):
            if r >= c:
                continue
            avg = (Y[r, c] + Y[c, r]) / 2.0
            scores[(int(r), int(c))] = float(avg)
        return scores

    def random_walk(
        self,
        adj: scipy.sparse.csr_matrix,
        *,
        walks_per_node: int = 32,
        return_chance: float = 0.3,
        expansions: int = 1,
        seed: int | None = None,
    ) -> dict[tuple[int, int], float]:
        """Compatibility alias for :meth:`walk_visit_expansion`."""
        return self.walk_visit_expansion(
            adj,
            walks_per_node=walks_per_node,
            return_chance=return_chance,
            expansions=expansions,
            seed=seed,
        )

    def local_walk_support(
        self,
        adj: scipy.sparse.csr_matrix,
        *,
        walk_len: int = 8,
        walks_per_edge: int = 1_000,
        gamma: float = 0.85,
        min_targets: int = 2,
        min_moves: int = 2,
        require_both: bool = False,
        agg: str = "mean",
        sample_size: int | None = None,
        sample_fraction: float | None = None,
        seed: int = 123,
        parallel: bool = True,
        n_threads: int | None = None,
    ) -> dict[tuple[int, int], float]:
        """Score candidate edges by local endpoint random-walk support. Low = suspect."""
        return _local_walk_support(
            adj,
            walk_len=walk_len,
            walks_per_edge=walks_per_edge,
            gamma=gamma,
            min_targets=min_targets,
            min_moves=min_moves,
            require_both=require_both,
            agg=agg,
            sample_size=sample_size,
            sample_fraction=sample_fraction,
            seed=seed,
            parallel=parallel,
            n_threads=n_threads,
        )

    def edge_support(
        self,
        adj: scipy.sparse.csr_matrix,
        *,
        walk_len: int = 8,
        walks_per_edge: int = 1_000,
        gamma: float = 0.85,
        min_targets: int = 2,
        min_moves: int = 2,
        require_both: bool = False,
        agg: str = "mean",
        sample_size: int | None = None,
        sample_fraction: float | None = None,
        seed: int = 123,
        parallel: bool = True,
        n_threads: int | None = None,
    ) -> dict[tuple[int, int], float]:
        """Compatibility alias for :meth:`local_walk_support`."""
        return self.local_walk_support(
            adj,
            walk_len=walk_len,
            walks_per_edge=walks_per_edge,
            gamma=gamma,
            min_targets=min_targets,
            min_moves=min_moves,
            require_both=require_both,
            agg=agg,
            sample_size=sample_size,
            sample_fraction=sample_fraction,
            seed=seed,
            parallel=parallel,
            n_threads=n_threads,
        )

Methods:

betweenness(adj)

Score edges by edge betweenness centrality.

High centrality signals a critical bridge — a characteristic of long-range false connections that span otherwise distinct communities.

Parameters:

Name	Type	Description	Default
adj	csr_matrix	Symmetric unweighted adjacency matrix.	required

Returns:

Type	Description
dict[tuple[int, int], float]	Edge → normalised betweenness in [0, 1]. High = suspect.

Notes

Uses networkx.edge_betweenness_centrality, runtime O(VE). Slow on graphs with more than a few thousand nodes.

Examples:

>>> from spatial_graph_algorithms.simulate import generate
>>> from spatial_graph_algorithms.denoise import EdgeScorer
>>> sg = generate(n=100, seed=0)
>>> scores = EdgeScorer().betweenness(sg.adjacency_matrix)
>>> all(0.0 <= v <= 1.0 for v in scores.values())
True

Source code in src/spatial_graph_algorithms/denoise/__init__.py

def betweenness(
    self,
    adj: scipy.sparse.csr_matrix,
) -> dict[tuple[int, int], float]:
    """Score edges by edge betweenness centrality.

    High centrality signals a critical bridge — a characteristic of
    long-range false connections that span otherwise distinct communities.

    Parameters
    ----------
    adj : scipy.sparse.csr_matrix
        Symmetric unweighted adjacency matrix.

    Returns
    -------
    dict[tuple[int, int], float]
        Edge → normalised betweenness in ``[0, 1]``.  High = suspect.

    Notes
    -----
    Uses ``networkx.edge_betweenness_centrality``, runtime ``O(VE)``.
    Slow on graphs with more than a few thousand nodes.

    Examples
    --------
    >>> from spatial_graph_algorithms.simulate import generate
    >>> from spatial_graph_algorithms.denoise import EdgeScorer
    >>> sg = generate(n=100, seed=0)
    >>> scores = EdgeScorer().betweenness(sg.adjacency_matrix)
    >>> all(0.0 <= v <= 1.0 for v in scores.values())
    True
    """
    G = nx.from_scipy_sparse_array(adj)
    centrality = nx.edge_betweenness_centrality(G, normalized=True)
    return {_canonical(u, v): s for (u, v), s in centrality.items()}

jaccard(adj)

Score edges by Jaccard neighbourhood overlap.

Edges connecting nodes with few shared neighbours are more likely to be false connections bridging unrelated graph regions.

Parameters:

Name	Type	Description	Default
adj	csr_matrix	Symmetric unweighted adjacency matrix for a unipartite graph.	required

Returns:

Type	Description
dict[tuple[int, int], float]	Edge → Jaccard score in [0, 1]. Low = suspect.

Raises:

Type	Description
ValueError	If adj represents a bipartite graph.

Examples:

>>> from spatial_graph_algorithms.simulate import generate
>>> from spatial_graph_algorithms.denoise import EdgeScorer
>>> sg = generate(n=100, seed=0)
>>> scores = EdgeScorer().jaccard(sg.adjacency_matrix)
>>> all(0.0 <= v <= 1.0 for v in scores.values())
True

Source code in src/spatial_graph_algorithms/denoise/__init__.py

def jaccard(
    self,
    adj: scipy.sparse.csr_matrix,
) -> dict[tuple[int, int], float]:
    """Score edges by Jaccard neighbourhood overlap.

    Edges connecting nodes with few shared neighbours are more likely to
    be false connections bridging unrelated graph regions.

    Parameters
    ----------
    adj : scipy.sparse.csr_matrix
        Symmetric unweighted adjacency matrix for a **unipartite** graph.

    Returns
    -------
    dict[tuple[int, int], float]
        Edge → Jaccard score in ``[0, 1]``.  Low = suspect.

    Raises
    ------
    ValueError
        If *adj* represents a bipartite graph.

    Examples
    --------
    >>> from spatial_graph_algorithms.simulate import generate
    >>> from spatial_graph_algorithms.denoise import EdgeScorer
    >>> sg = generate(n=100, seed=0)
    >>> scores = EdgeScorer().jaccard(sg.adjacency_matrix)
    >>> all(0.0 <= v <= 1.0 for v in scores.values())
    True
    """
    G = nx.from_scipy_sparse_array(adj)
    if nx.is_bipartite(G):
        raise ValueError(
            "jaccard scorer is designed for unipartite graphs. "
            "Use square_bipartite for bipartite graphs."
        )
    scores: dict[tuple[int, int], float] = {}
    rows, cols = adj.nonzero()
    for r, c in zip(rows, cols):
        if r >= c:
            continue
        nb_u = set(adj[r].indices)
        nb_v = set(adj[c].indices)
        union = len(nb_u | nb_v)
        scores[(int(r), int(c))] = len(nb_u & nb_v) / union if union > 0 else 0.0
    return scores

square_bipartite(adj)

Score edges by 4-cycle participation in a bipartite graph.

True edges in a spatial bipartite graph typically close many 4-cycles (shared neighbours in the opposite partition). False edges span unrelated partitions and form fewer squares.

Parameters:

Name	Type	Description	Default
adj	csr_matrix	Symmetric unweighted adjacency matrix for a bipartite graph.	required

Returns:

Type	Description
dict[tuple[int, int], float]	Edge → 4-cycle count (as float). Low = suspect.

Raises:

Type	Description
ValueError	If adj does not represent a bipartite graph.

Examples:

>>> from spatial_graph_algorithms.simulate import generate
>>> from spatial_graph_algorithms.denoise import EdgeScorer
>>> sg = generate(n=80, mode="knn_bipartite", seed=0)
>>> scores = EdgeScorer().square_bipartite(sg.adjacency_matrix)
>>> all(v >= 0 for v in scores.values())
True

Source code in src/spatial_graph_algorithms/denoise/__init__.py

def square_bipartite(
    self,
    adj: scipy.sparse.csr_matrix,
) -> dict[tuple[int, int], float]:
    """Score edges by 4-cycle participation in a bipartite graph.

    True edges in a spatial bipartite graph typically close many 4-cycles
    (shared neighbours in the opposite partition).  False edges span
    unrelated partitions and form fewer squares.

    Parameters
    ----------
    adj : scipy.sparse.csr_matrix
        Symmetric unweighted adjacency matrix for a **bipartite** graph.

    Returns
    -------
    dict[tuple[int, int], float]
        Edge → 4-cycle count (as float).  Low = suspect.

    Raises
    ------
    ValueError
        If *adj* does not represent a bipartite graph.

    Examples
    --------
    >>> from spatial_graph_algorithms.simulate import generate
    >>> from spatial_graph_algorithms.denoise import EdgeScorer
    >>> sg = generate(n=80, mode="knn_bipartite", seed=0)
    >>> scores = EdgeScorer().square_bipartite(sg.adjacency_matrix)
    >>> all(v >= 0 for v in scores.values())
    True
    """
    G = nx.from_scipy_sparse_array(adj)
    if not nx.is_bipartite(G):
        raise ValueError(
            "square_bipartite scorer requires a bipartite graph. "
            "Use jaccard for unipartite graphs."
        )
    top_nodes, bottom_nodes = nx.bipartite.sets(G)
    top_adj = {u: set(G[u]) for u in top_nodes}
    bottom_adj = {v: set(G[v]) for v in bottom_nodes}

    edge_scores: dict[tuple[int, int], float] = {}
    for u, v in G.edges():
        if u in bottom_nodes:
            u, v = v, u  # ensure u ∈ top, v ∈ bottom
        b_u = top_adj[u] - {v}
        t_v = bottom_adj[v] - {u}
        count = sum(len(top_adj[w].intersection(b_u)) for w in t_v)
        edge_scores[_canonical(u, v)] = float(count)
    return edge_scores

square_bipartite_linalg(adj)

Alternative square_bipartite implementation using sparse matrix multiplication.

Source code in src/spatial_graph_algorithms/denoise/__init__.py

def square_bipartite_linalg(
    self,
    adj: scipy.sparse.csr_matrix,
) -> dict[tuple[int, int], float]:
    """Alternative square_bipartite implementation using sparse matrix multiplication."""
    # W3 = A @ A @ A - diag(A @ A) @ A - A @ diag(A @ A) + A.multiply(A).multiply(A)
    # (total 3-paths) - (i->k->i->j) - (i->j->k->j) + (i->j->i->j)
    adj      = adj.astype(np.int64)

    adj2     = adj @ adj
    adj2_diag   = adj2.diagonal().astype(np.int64)

    adj3     = adj2 @ adj

    w3 = (
        adj3
        - scipy.sparse.diags(adj2_diag, format="csr", dtype=np.int64) @ adj
        - adj @ scipy.sparse.diags(adj2_diag, format="csr", dtype=np.int64)
        + adj.multiply(adj).multiply(adj)
    )

    # only take scores if there is an edge in the original graph
    w3 = w3.multiply(adj > 0)

    # align w3 values with adj edges
    scores: dict[tuple[int, int], float] = {}
    rows, cols = adj.nonzero()
    for r, c in zip(rows, cols):
        scores[(int(r), int(c))] = float(w3[r, c])
    return scores

community(adj, *, n_runs=5)

Score edges by Louvain community membership consistency.

Runs the Louvain algorithm (via igraph) n_runs times. Each edge is scored by the fraction of runs it was classified as intra-community. Edges that persistently bridge communities are likely false connections.

Parameters:

Name	Type	Description	Default
adj	csr_matrix	Symmetric unweighted adjacency matrix.	required
n_runs	int	Independent Louvain repetitions. Higher = more stable scores. Default 5.	5

Returns:

Type	Description
dict[tuple[int, int], float]	Edge → intra-community fraction in [0, 1]. Low = suspect.

Examples:

>>> from spatial_graph_algorithms.simulate import generate
>>> from spatial_graph_algorithms.denoise import EdgeScorer
>>> sg = generate(n=100, seed=0)
>>> scores = EdgeScorer().community(sg.adjacency_matrix, n_runs=3)
>>> all(0.0 <= v <= 1.0 for v in scores.values())
True

Source code in src/spatial_graph_algorithms/denoise/__init__.py

def community(
    self,
    adj: scipy.sparse.csr_matrix,
    *,
    n_runs: int = 5,
) -> dict[tuple[int, int], float]:
    """Score edges by Louvain community membership consistency.

    Runs the Louvain algorithm (via igraph) *n_runs* times.  Each edge
    is scored by the fraction of runs it was classified as intra-community.
    Edges that persistently bridge communities are likely false connections.

    Parameters
    ----------
    adj : scipy.sparse.csr_matrix
        Symmetric unweighted adjacency matrix.
    n_runs : int
        Independent Louvain repetitions.  Higher = more stable scores.
        Default 5.

    Returns
    -------
    dict[tuple[int, int], float]
        Edge → intra-community fraction in ``[0, 1]``.  Low = suspect.

    Examples
    --------
    >>> from spatial_graph_algorithms.simulate import generate
    >>> from spatial_graph_algorithms.denoise import EdgeScorer
    >>> sg = generate(n=100, seed=0)
    >>> scores = EdgeScorer().community(sg.adjacency_matrix, n_runs=3)
    >>> all(0.0 <= v <= 1.0 for v in scores.values())
    True
    """
    rows, cols = adj.nonzero()
    ig_edges = [(int(r), int(c)) for r, c in zip(rows, cols) if r < c]
    G_ig = _igraph.Graph(n=adj.shape[0], edges=ig_edges, directed=False)

    canonical_edges = {
        (int(min(r, c)), int(max(r, c)))
        for r, c in zip(rows, cols)
        if r != c
    }
    within: dict[tuple[int, int], int] = {e: 0 for e in canonical_edges}

    for _ in range(n_runs):
        partition = G_ig.community_multilevel()
        membership = partition.membership
        for e in canonical_edges:
            if membership[e[0]] == membership[e[1]]:
                within[e] += 1

    return {e: cnt / n_runs for e, cnt in within.items()}

walk_visit_expansion(adj, *, walks_per_node=32, return_chance=0.3, expansions=1, seed=None)

Score edges by bridge-filtered walk visitation expansion.

Performs walks_per_node random walks from every node. At each step there is a return_chance of attempting to return home. A walk only counts when the current position has two edge-disjoint paths back to the origin in the locally discovered subgraph — edges that act as bridges (typical of false connections) will have their walks discarded more often, yielding lower scores.

After normalisation, the visitation matrix is raised to the power 2 ** expansions via repeated matrix squaring (expansions = 1 computes Y @ Y), amplifying score separation between true and false edges.

Parameters:

Name	Type	Description	Default
adj	csr_matrix	Symmetric unweighted adjacency matrix.	required
walks_per_node	int	Walks launched per origin node. Default 32.	32
return_chance	float	Per-step probability of attempting a return. Default 0.3.	0.3
expansions	int	Number of matrix-squaring steps applied after normalisation. Default 1 (computes Y @ Y). Set to 0 to skip.	1
seed	int	Random seed for reproducibility.	None

Returns:

Type	Description
dict[tuple[int, int], float]	Edge → normalised visitation frequency. Low = suspect.

Warns:

Type	Description
ResourceWarning	When adj has more than 2 000 nodes (O(n²) memory footprint).

Examples:

>>> from spatial_graph_algorithms.simulate import generate
>>> from spatial_graph_algorithms.denoise import EdgeScorer
>>> sg = generate(n=60, seed=0)
>>> scores = EdgeScorer().walk_visit_expansion(
...     sg.adjacency_matrix, walks_per_node=4, seed=0
... )
>>> len(scores) == sg.n_edges
True

Source code in src/spatial_graph_algorithms/denoise/__init__.py

def walk_visit_expansion(
    self,
    adj: scipy.sparse.csr_matrix,
    *,
    walks_per_node: int = 32,
    return_chance: float = 0.3,
    expansions: int = 1,
    seed: int | None = None,
) -> dict[tuple[int, int], float]:
    """Score edges by bridge-filtered walk visitation expansion.

    Performs *walks_per_node* random walks from every node.  At each step
    there is a *return_chance* of attempting to return home.  A walk only
    counts when the current position has **two edge-disjoint paths** back
    to the origin in the locally discovered subgraph — edges that act as
    bridges (typical of false connections) will have their walks discarded
    more often, yielding lower scores.

    After normalisation, the visitation matrix is raised to the power
    ``2 ** expansions`` via repeated matrix squaring (*expansions* = 1
    computes ``Y @ Y``), amplifying score separation between true and
    false edges.

    Parameters
    ----------
    adj : scipy.sparse.csr_matrix
        Symmetric unweighted adjacency matrix.
    walks_per_node : int
        Walks launched per origin node.  Default 32.
    return_chance : float
        Per-step probability of attempting a return.  Default 0.3.
    expansions : int
        Number of matrix-squaring steps applied after normalisation.
        Default 1 (computes ``Y @ Y``).  Set to 0 to skip.
    seed : int, optional
        Random seed for reproducibility.

    Returns
    -------
    dict[tuple[int, int], float]
        Edge → normalised visitation frequency.  Low = suspect.

    Warns
    -----
    ResourceWarning
        When *adj* has more than 2 000 nodes (O(n²) memory footprint).

    Examples
    --------
    >>> from spatial_graph_algorithms.simulate import generate
    >>> from spatial_graph_algorithms.denoise import EdgeScorer
    >>> sg = generate(n=60, seed=0)
    >>> scores = EdgeScorer().walk_visit_expansion(
    ...     sg.adjacency_matrix, walks_per_node=4, seed=0
    ... )
    >>> len(scores) == sg.n_edges
    True
    """
    import warnings

    n = adj.shape[0]
    if n > 2000:
        warnings.warn(
            f"walk_visit_expansion scorer allocates a ({n}×{n}) visitation matrix "
            f"({n * n * 8 / 1e6:.0f} MB). "
            "Consider betweenness or jaccard for large graphs.",
            ResourceWarning,
            stacklevel=2,
        )

    rng = np.random.default_rng(seed)
    G = nx.from_scipy_sparse_array(adj)
    adj_list = [list(G.neighbors(i)) for i in range(n)]
    visit = np.zeros((n, n), dtype=np.float64)

    for origin in range(n):
        for _ in range(walks_per_node):
            current = origin
            path: list[tuple[int, int]] = []
            discovered = nx.Graph()
            discovered.add_node(origin)

            while True:
                nbs = adj_list[current]
                if not nbs:
                    break
                nxt = int(rng.choice(nbs))
                path.append((current, nxt))

                discovered.add_edge(current, nxt, weight=1)

                # 1-hop BFS to enrich the discovered subgraph (matches
                # the original's mini-BFS with range(1))
                for nb in adj_list[nxt]:
                    if not discovered.has_edge(nxt, nb):
                        discovered.add_edge(nxt, nb, weight=1)

                if rng.random() < return_chance:
                    # A walk only counts when two edge-disjoint paths exist
                    # from the current position back to the origin.
                    # Find the first path, remove its edges, then verify a
                    # second path still exists — this selectively discards
                    # walks that traversed bridges or near-bridges.
                    temp = discovered.copy()
                    try:
                        first = nx.shortest_path(temp, source=current,
                                                 target=origin,
                                                 weight="weight")
                        for k in range(len(first) - 1):
                            temp.remove_edge(first[k], first[k + 1])
                        nx.shortest_path(temp, source=current,
                                         target=origin, weight="weight")
                    except nx.NetworkXNoPath:
                        path = []
                    break

                current = nxt

            for src, dst in path:
                visit[origin, dst] += 1
                visit[origin, src] += 1

    # Row-normalise
    row_sums = visit.sum(axis=1)
    with np.errstate(divide="ignore", invalid="ignore"):
        norm_factors = np.where(row_sums > 0, 1.0 / row_sums, 0.0)
    Y = visit * norm_factors[:, np.newaxis]

    # Matrix-squaring expansion: amplifies score separation
    for _ in range(expansions):
        Y = np.dot(Y, Y)

    scores: dict[tuple[int, int], float] = {}
    rows, cols = adj.nonzero()
    for r, c in zip(rows, cols):
        if r >= c:
            continue
        avg = (Y[r, c] + Y[c, r]) / 2.0
        scores[(int(r), int(c))] = float(avg)
    return scores

random_walk(adj, *, walks_per_node=32, return_chance=0.3, expansions=1, seed=None)

Compatibility alias for :meth:walk_visit_expansion.

Source code in src/spatial_graph_algorithms/denoise/__init__.py

def random_walk(
    self,
    adj: scipy.sparse.csr_matrix,
    *,
    walks_per_node: int = 32,
    return_chance: float = 0.3,
    expansions: int = 1,
    seed: int | None = None,
) -> dict[tuple[int, int], float]:
    """Compatibility alias for :meth:`walk_visit_expansion`."""
    return self.walk_visit_expansion(
        adj,
        walks_per_node=walks_per_node,
        return_chance=return_chance,
        expansions=expansions,
        seed=seed,
    )

local_walk_support(adj, *, walk_len=8, walks_per_edge=1000, gamma=0.85, min_targets=2, min_moves=2, require_both=False, agg='mean', sample_size=None, sample_fraction=None, seed=123, parallel=True, n_threads=None)

Score candidate edges by local endpoint random-walk support. Low = suspect.

Source code in src/spatial_graph_algorithms/denoise/__init__.py

def local_walk_support(
    self,
    adj: scipy.sparse.csr_matrix,
    *,
    walk_len: int = 8,
    walks_per_edge: int = 1_000,
    gamma: float = 0.85,
    min_targets: int = 2,
    min_moves: int = 2,
    require_both: bool = False,
    agg: str = "mean",
    sample_size: int | None = None,
    sample_fraction: float | None = None,
    seed: int = 123,
    parallel: bool = True,
    n_threads: int | None = None,
) -> dict[tuple[int, int], float]:
    """Score candidate edges by local endpoint random-walk support. Low = suspect."""
    return _local_walk_support(
        adj,
        walk_len=walk_len,
        walks_per_edge=walks_per_edge,
        gamma=gamma,
        min_targets=min_targets,
        min_moves=min_moves,
        require_both=require_both,
        agg=agg,
        sample_size=sample_size,
        sample_fraction=sample_fraction,
        seed=seed,
        parallel=parallel,
        n_threads=n_threads,
    )

edge_support(adj, *, walk_len=8, walks_per_edge=1000, gamma=0.85, min_targets=2, min_moves=2, require_both=False, agg='mean', sample_size=None, sample_fraction=None, seed=123, parallel=True, n_threads=None)

Compatibility alias for :meth:local_walk_support.

Source code in src/spatial_graph_algorithms/denoise/__init__.py

def edge_support(
    self,
    adj: scipy.sparse.csr_matrix,
    *,
    walk_len: int = 8,
    walks_per_edge: int = 1_000,
    gamma: float = 0.85,
    min_targets: int = 2,
    min_moves: int = 2,
    require_both: bool = False,
    agg: str = "mean",
    sample_size: int | None = None,
    sample_fraction: float | None = None,
    seed: int = 123,
    parallel: bool = True,
    n_threads: int | None = None,
) -> dict[tuple[int, int], float]:
    """Compatibility alias for :meth:`local_walk_support`."""
    return self.local_walk_support(
        adj,
        walk_len=walk_len,
        walks_per_edge=walks_per_edge,
        gamma=gamma,
        min_targets=min_targets,
        min_moves=min_moves,
        require_both=require_both,
        agg=agg,
        sample_size=sample_size,
        sample_fraction=sample_fraction,
        seed=seed,
        parallel=parallel,
        n_threads=n_threads,
    )

spatial_graph_algorithms.denoise.EdgeFilterer

Remove edges from a SpatialGraph using scored edge rankings.

All methods accept a :class:SpatialGraph and a scores dict (from :class:EdgeScorer) and return a new SpatialGraph with:

• Fewer edges in the adjacency matrix.
• edge_metadata["is_removed"] boolean column (created fresh if edge_metadata was None).
• The largest connected component extracted — nodes that lose all edges after filtering are removed so the returned graph is always connected.

Methods:

Name	Description
by_fraction	Remove the worst fraction of scored edges.
by_threshold	Remove edges above or below an explicit score value.
by_percentile	Remove edges outside a percentile band.
by_optimal_f1	Oracle: sweep thresholds, pick the one maximising F1 vs ground truth.

Source code in src/spatial_graph_algorithms/denoise/__init__.py

class EdgeFilterer:
    """Remove edges from a SpatialGraph using scored edge rankings.

    All methods accept a :class:`SpatialGraph` and a *scores* dict (from
    :class:`EdgeScorer`) and return a new ``SpatialGraph`` with:

    - Fewer edges in the adjacency matrix.
    - ``edge_metadata["is_removed"]`` boolean column (created fresh if
      ``edge_metadata`` was ``None``).
    - The largest connected component extracted — nodes that lose all edges
      after filtering are removed so the returned graph is always connected.

    Methods
    -------
    by_fraction(sg, scores, *, fraction, method)
        Remove the worst *fraction* of scored edges.
    by_threshold(sg, scores, *, threshold, method)
        Remove edges above or below an explicit score value.
    by_percentile(sg, scores, *, percentile, method)
        Remove edges outside a percentile band.
    by_optimal_f1(sg, scores)
        Oracle: sweep thresholds, pick the one maximising F1 vs ground truth.
    """

    def _apply(
        self,
        sg: SpatialGraph,
        removed_edges: set[tuple[int, int]],
    ) -> SpatialGraph:
        new_adj = _build_new_adjacency(sg.adjacency_matrix, removed_edges)
        new_meta = _build_new_metadata(sg, removed_edges)
        # Build without edge_metadata so _extract_lcc doesn't filter the audit trail.
        # edge_metadata records all original edges (including removed ones); it is
        # set after construction to preserve the full denoising history for evaluate_denoising().
        sg_clean = SpatialGraph(
            adjacency_matrix=new_adj,
            positions=sg.positions,
            reconstructed_positions=sg.reconstructed_positions,
            node_metadata=sg.node_metadata,
            node_id_map=sg.node_id_map,
            keep_lcc=True,
        )
        sg_clean.edge_metadata = new_meta
        return sg_clean

    def by_fraction(
        self,
        sg: SpatialGraph,
        scores: dict[tuple[int, int], float],
        *,
        fraction: float,
        method: str,
    ) -> SpatialGraph:
        """Remove the worst *fraction* of scored edges.

        Parameters
        ----------
        sg : SpatialGraph
            Input graph.
        scores : dict
            Edge scores from :class:`EdgeScorer`.
        fraction : float
            Fraction of edges to remove, in ``[0, 1]``.
        method : str
            Scoring method that produced *scores*.  Score polarity is looked
            up in :data:`SCORE_POLARITY`.

        Returns
        -------
        SpatialGraph
            New graph with edges removed and ``is_removed`` metadata set.

        Raises
        ------
        ValueError
            If *fraction* is not in ``[0, 1]`` or *method* is unknown.

        Examples
        --------
        >>> from spatial_graph_algorithms.simulate import generate
        >>> from spatial_graph_algorithms.denoise import EdgeScorer, EdgeFilterer
        >>> sg = generate(n=100, false_edge_fraction=0.1, seed=0)
        >>> scores = EdgeScorer().jaccard(sg.adjacency_matrix)
        >>> sg_clean = EdgeFilterer().by_fraction(
        ...     sg, scores, fraction=0.05, method="jaccard"
        ... )
        >>> sg_clean.n_edges < sg.n_edges
        True
        """
        if not 0.0 <= fraction <= 1.0:
            raise ValueError(f"fraction must be in [0, 1], got {fraction}")
        remove_high_scores = _score_direction(method)
        sorted_edges = sorted(scores.items(), key=lambda x: x[1], reverse=remove_high_scores)
        n_remove = int(len(sorted_edges) * fraction)
        removed = {e for e, _ in sorted_edges[:n_remove]}
        return self._apply(sg, removed)

    def by_threshold(
        self,
        sg: SpatialGraph,
        scores: dict[tuple[int, int], float],
        *,
        threshold: float,
        method: str,
    ) -> SpatialGraph:
        """Remove edges whose score crosses *threshold*.

        Parameters
        ----------
        sg : SpatialGraph
            Input graph.
        scores : dict
            Edge scores from :class:`EdgeScorer`.
        threshold : float
            Score cut-off value.
        method : str
            Scoring method that produced *scores*.  Score polarity is looked
            up in :data:`SCORE_POLARITY`.

        Returns
        -------
        SpatialGraph
            New graph with edges removed and ``is_removed`` metadata set.

        Examples
        --------
        >>> sg_clean = EdgeFilterer().by_threshold(
        ...     sg, scores, threshold=0.5, method="betweenness"
        ... )
        """
        return self._by_threshold_direction(
            sg, scores, threshold=threshold, remove_high_scores=_score_direction(method)
        )

    def _by_threshold_direction(
        self,
        sg: SpatialGraph,
        scores: dict[tuple[int, int], float],
        *,
        threshold: float,
        remove_high_scores: bool,
    ) -> SpatialGraph:
        """Remove edges above or below *threshold* using a resolved score direction."""
        if remove_high_scores:
            removed = {e for e, s in scores.items() if s >= threshold}
        else:
            removed = {e for e, s in scores.items() if s <= threshold}
        return self._apply(sg, removed)

    def by_percentile(
        self,
        sg: SpatialGraph,
        scores: dict[tuple[int, int], float],
        *,
        percentile: float,
        method: str,
    ) -> SpatialGraph:
        """Remove edges outside a percentile band.

        Parameters
        ----------
        sg : SpatialGraph
            Input graph.
        scores : dict
            Edge scores from :class:`EdgeScorer`.
        percentile : float
            Percentage of edges to remove from the worst end, in ``[0, 100]``.
        method : str
            Scoring method that produced *scores*.  Score polarity is looked
            up in :data:`SCORE_POLARITY`.

        Returns
        -------
        SpatialGraph
            New graph with edges removed and ``is_removed`` metadata set.

        Raises
        ------
        ValueError
            If *percentile* is not in ``[0, 100]`` or *method* is unknown.

        Examples
        --------
        >>> sg_clean = EdgeFilterer().by_percentile(
        ...     sg, scores, percentile=5.0, method="jaccard"
        ... )
        """
        if not 0.0 <= percentile <= 100.0:
            raise ValueError(f"percentile must be in [0, 100], got {percentile}")
        vals = np.array(list(scores.values()))
        remove_high_scores = _score_direction(method)
        if remove_high_scores:
            threshold = float(np.percentile(vals, 100.0 - percentile))
            removed = {e for e, s in scores.items() if s >= threshold}
        else:
            threshold = float(np.percentile(vals, percentile))
            removed = {e for e, s in scores.items() if s <= threshold}
        return self._apply(sg, removed)

    def by_optimal_f1(
        self,
        sg: SpatialGraph,
        scores: dict[tuple[int, int], float],
    ) -> SpatialGraph:
        """Oracle filter: remove edges at the threshold that maximises F1.

        Sweeps every candidate threshold and selects the one that maximises
        the F1 score between removed edges and the ground-truth ``is_false``
        labels.  Both score polarities are tried; the globally best threshold
        wins regardless of direction.

        This is an **oracle method** — it requires ground-truth labels and
        therefore only applies to simulated graphs.  Use it to establish
        the theoretical upper bound for a given scorer.

        Parameters
        ----------
        sg : SpatialGraph
            Input graph.  Must have ``edge_metadata["is_false"]``.
        scores : dict
            Edge scores from :class:`EdgeScorer`.

        Returns
        -------
        SpatialGraph
            Graph filtered at the optimal F1 threshold, with ``is_removed``
            metadata set.

        Raises
        ------
        ValueError
            If ``sg.edge_metadata`` is ``None`` or lacks an ``is_false`` column.

        Examples
        --------
        >>> from spatial_graph_algorithms.simulate import generate
        >>> from spatial_graph_algorithms.denoise import EdgeScorer, EdgeFilterer
        >>> from spatial_graph_algorithms.metrics import evaluate_denoising
        >>> sg = generate(n=200, false_edge_fraction=0.10, seed=42)
        >>> scores = EdgeScorer().jaccard(sg.adjacency_matrix)
        >>> sg_opt = EdgeFilterer().by_optimal_f1(sg, scores)
        >>> report = evaluate_denoising(sg_opt)
        >>> report["f1"] >= 0.0
        True
        """
        if sg.edge_metadata is None or "is_false" not in sg.edge_metadata.columns:
            raise ValueError(
                "by_optimal_f1 requires ground-truth labels: "
                "sg.edge_metadata must have an 'is_false' column."
            )
        meta = sg.edge_metadata
        lo = np.minimum(meta["source"].values, meta["target"].values).astype(np.int64)
        hi = np.maximum(meta["source"].values, meta["target"].values).astype(np.int64)
        is_false_vals = meta["is_false"].values
        labels: dict[tuple[int, int], bool] = {
            (int(lo[i]), int(hi[i])): bool(is_false_vals[i])
            for i in range(len(meta))
        }
        threshold, _best_f1, remove_high_scores = _find_optimal_f1_threshold(scores, labels)
        return self._by_threshold_direction(
            sg, scores, threshold=threshold, remove_high_scores=remove_high_scores
        )

Methods:

by_fraction(sg, scores, *, fraction, method)

Remove the worst fraction of scored edges.

Parameters:

Name	Type	Description	Default
sg	SpatialGraph	Input graph.	required
scores	dict	Edge scores from :class:EdgeScorer.	required
fraction	float	Fraction of edges to remove, in [0, 1].	required
method	str	Scoring method that produced scores. Score polarity is looked up in :data:SCORE_POLARITY.	required

Returns:

Type	Description
SpatialGraph	New graph with edges removed and is_removed metadata set.

Raises:

Type	Description
ValueError	If fraction is not in [0, 1] or method is unknown.

Examples:

>>> from spatial_graph_algorithms.simulate import generate
>>> from spatial_graph_algorithms.denoise import EdgeScorer, EdgeFilterer
>>> sg = generate(n=100, false_edge_fraction=0.1, seed=0)
>>> scores = EdgeScorer().jaccard(sg.adjacency_matrix)
>>> sg_clean = EdgeFilterer().by_fraction(
...     sg, scores, fraction=0.05, method="jaccard"
... )
>>> sg_clean.n_edges < sg.n_edges
True

Source code in src/spatial_graph_algorithms/denoise/__init__.py

def by_fraction(
    self,
    sg: SpatialGraph,
    scores: dict[tuple[int, int], float],
    *,
    fraction: float,
    method: str,
) -> SpatialGraph:
    """Remove the worst *fraction* of scored edges.

    Parameters
    ----------
    sg : SpatialGraph
        Input graph.
    scores : dict
        Edge scores from :class:`EdgeScorer`.
    fraction : float
        Fraction of edges to remove, in ``[0, 1]``.
    method : str
        Scoring method that produced *scores*.  Score polarity is looked
        up in :data:`SCORE_POLARITY`.

    Returns
    -------
    SpatialGraph
        New graph with edges removed and ``is_removed`` metadata set.

    Raises
    ------
    ValueError
        If *fraction* is not in ``[0, 1]`` or *method* is unknown.

    Examples
    --------
    >>> from spatial_graph_algorithms.simulate import generate
    >>> from spatial_graph_algorithms.denoise import EdgeScorer, EdgeFilterer
    >>> sg = generate(n=100, false_edge_fraction=0.1, seed=0)
    >>> scores = EdgeScorer().jaccard(sg.adjacency_matrix)
    >>> sg_clean = EdgeFilterer().by_fraction(
    ...     sg, scores, fraction=0.05, method="jaccard"
    ... )
    >>> sg_clean.n_edges < sg.n_edges
    True
    """
    if not 0.0 <= fraction <= 1.0:
        raise ValueError(f"fraction must be in [0, 1], got {fraction}")
    remove_high_scores = _score_direction(method)
    sorted_edges = sorted(scores.items(), key=lambda x: x[1], reverse=remove_high_scores)
    n_remove = int(len(sorted_edges) * fraction)
    removed = {e for e, _ in sorted_edges[:n_remove]}
    return self._apply(sg, removed)

by_threshold(sg, scores, *, threshold, method)

Remove edges whose score crosses threshold.

Parameters:

Name	Type	Description	Default
sg	SpatialGraph	Input graph.	required
scores	dict	Edge scores from :class:EdgeScorer.	required
threshold	float	Score cut-off value.	required
method	str	Scoring method that produced scores. Score polarity is looked up in :data:SCORE_POLARITY.	required

Returns:

Type	Description
SpatialGraph	New graph with edges removed and is_removed metadata set.

Examples:

>>> sg_clean = EdgeFilterer().by_threshold(
...     sg, scores, threshold=0.5, method="betweenness"
... )

Source code in src/spatial_graph_algorithms/denoise/__init__.py

def by_threshold(
    self,
    sg: SpatialGraph,
    scores: dict[tuple[int, int], float],
    *,
    threshold: float,
    method: str,
) -> SpatialGraph:
    """Remove edges whose score crosses *threshold*.

    Parameters
    ----------
    sg : SpatialGraph
        Input graph.
    scores : dict
        Edge scores from :class:`EdgeScorer`.
    threshold : float
        Score cut-off value.
    method : str
        Scoring method that produced *scores*.  Score polarity is looked
        up in :data:`SCORE_POLARITY`.

    Returns
    -------
    SpatialGraph
        New graph with edges removed and ``is_removed`` metadata set.

    Examples
    --------
    >>> sg_clean = EdgeFilterer().by_threshold(
    ...     sg, scores, threshold=0.5, method="betweenness"
    ... )
    """
    return self._by_threshold_direction(
        sg, scores, threshold=threshold, remove_high_scores=_score_direction(method)
    )

by_percentile(sg, scores, *, percentile, method)

Remove edges outside a percentile band.

Parameters:

Name	Type	Description	Default
sg	SpatialGraph	Input graph.	required
scores	dict	Edge scores from :class:EdgeScorer.	required
percentile	float	Percentage of edges to remove from the worst end, in [0, 100].	required
method	str	Scoring method that produced scores. Score polarity is looked up in :data:SCORE_POLARITY.	required

Returns:

Type	Description
SpatialGraph	New graph with edges removed and is_removed metadata set.

Raises:

Type	Description
ValueError	If percentile is not in [0, 100] or method is unknown.

Examples:

>>> sg_clean = EdgeFilterer().by_percentile(
...     sg, scores, percentile=5.0, method="jaccard"
... )

Source code in src/spatial_graph_algorithms/denoise/__init__.py

def by_percentile(
    self,
    sg: SpatialGraph,
    scores: dict[tuple[int, int], float],
    *,
    percentile: float,
    method: str,
) -> SpatialGraph:
    """Remove edges outside a percentile band.

    Parameters
    ----------
    sg : SpatialGraph
        Input graph.
    scores : dict
        Edge scores from :class:`EdgeScorer`.
    percentile : float
        Percentage of edges to remove from the worst end, in ``[0, 100]``.
    method : str
        Scoring method that produced *scores*.  Score polarity is looked
        up in :data:`SCORE_POLARITY`.

    Returns
    -------
    SpatialGraph
        New graph with edges removed and ``is_removed`` metadata set.

    Raises
    ------
    ValueError
        If *percentile* is not in ``[0, 100]`` or *method* is unknown.

    Examples
    --------
    >>> sg_clean = EdgeFilterer().by_percentile(
    ...     sg, scores, percentile=5.0, method="jaccard"
    ... )
    """
    if not 0.0 <= percentile <= 100.0:
        raise ValueError(f"percentile must be in [0, 100], got {percentile}")
    vals = np.array(list(scores.values()))
    remove_high_scores = _score_direction(method)
    if remove_high_scores:
        threshold = float(np.percentile(vals, 100.0 - percentile))
        removed = {e for e, s in scores.items() if s >= threshold}
    else:
        threshold = float(np.percentile(vals, percentile))
        removed = {e for e, s in scores.items() if s <= threshold}
    return self._apply(sg, removed)

by_optimal_f1(sg, scores)

Oracle filter: remove edges at the threshold that maximises F1.

Sweeps every candidate threshold and selects the one that maximises the F1 score between removed edges and the ground-truth is_false labels. Both score polarities are tried; the globally best threshold wins regardless of direction.

This is an oracle method — it requires ground-truth labels and therefore only applies to simulated graphs. Use it to establish the theoretical upper bound for a given scorer.

Parameters:

Name	Type	Description	Default
sg	SpatialGraph	Input graph. Must have edge_metadata["is_false"].	required
scores	dict	Edge scores from :class:EdgeScorer.	required

Returns:

Type	Description
SpatialGraph	Graph filtered at the optimal F1 threshold, with is_removed metadata set.

Raises:

Type	Description
ValueError	If sg.edge_metadata is None or lacks an is_false column.

Examples:

>>> from spatial_graph_algorithms.simulate import generate
>>> from spatial_graph_algorithms.denoise import EdgeScorer, EdgeFilterer
>>> from spatial_graph_algorithms.metrics import evaluate_denoising
>>> sg = generate(n=200, false_edge_fraction=0.10, seed=42)
>>> scores = EdgeScorer().jaccard(sg.adjacency_matrix)
>>> sg_opt = EdgeFilterer().by_optimal_f1(sg, scores)
>>> report = evaluate_denoising(sg_opt)
>>> report["f1"] >= 0.0
True

Source code in src/spatial_graph_algorithms/denoise/__init__.py

def by_optimal_f1(
    self,
    sg: SpatialGraph,
    scores: dict[tuple[int, int], float],
) -> SpatialGraph:
    """Oracle filter: remove edges at the threshold that maximises F1.

    Sweeps every candidate threshold and selects the one that maximises
    the F1 score between removed edges and the ground-truth ``is_false``
    labels.  Both score polarities are tried; the globally best threshold
    wins regardless of direction.

    This is an **oracle method** — it requires ground-truth labels and
    therefore only applies to simulated graphs.  Use it to establish
    the theoretical upper bound for a given scorer.

    Parameters
    ----------
    sg : SpatialGraph
        Input graph.  Must have ``edge_metadata["is_false"]``.
    scores : dict
        Edge scores from :class:`EdgeScorer`.

    Returns
    -------
    SpatialGraph
        Graph filtered at the optimal F1 threshold, with ``is_removed``
        metadata set.

    Raises
    ------
    ValueError
        If ``sg.edge_metadata`` is ``None`` or lacks an ``is_false`` column.

    Examples
    --------
    >>> from spatial_graph_algorithms.simulate import generate
    >>> from spatial_graph_algorithms.denoise import EdgeScorer, EdgeFilterer
    >>> from spatial_graph_algorithms.metrics import evaluate_denoising
    >>> sg = generate(n=200, false_edge_fraction=0.10, seed=42)
    >>> scores = EdgeScorer().jaccard(sg.adjacency_matrix)
    >>> sg_opt = EdgeFilterer().by_optimal_f1(sg, scores)
    >>> report = evaluate_denoising(sg_opt)
    >>> report["f1"] >= 0.0
    True
    """
    if sg.edge_metadata is None or "is_false" not in sg.edge_metadata.columns:
        raise ValueError(
            "by_optimal_f1 requires ground-truth labels: "
            "sg.edge_metadata must have an 'is_false' column."
        )
    meta = sg.edge_metadata
    lo = np.minimum(meta["source"].values, meta["target"].values).astype(np.int64)
    hi = np.maximum(meta["source"].values, meta["target"].values).astype(np.int64)
    is_false_vals = meta["is_false"].values
    labels: dict[tuple[int, int], bool] = {
        (int(lo[i]), int(hi[i])): bool(is_false_vals[i])
        for i in range(len(meta))
    }
    threshold, _best_f1, remove_high_scores = _find_optimal_f1_threshold(scores, labels)
    return self._by_threshold_direction(
        sg, scores, threshold=threshold, remove_high_scores=remove_high_scores
    )

spatial_graph_algorithms.denoise.score_edges(sg, *, method, **kwargs)

Score all edges in sg using the specified method.

Convenience wrapper around :class:EdgeScorer.

Parameters:

Name	Type	Description	Default
sg	SpatialGraph	Input graph.	required
method	str	One of "betweenness", "jaccard", "square_bipartite", "community", "walk_visit_expansion", or "local_walk_support".	required
**kwargs	object	Forwarded to the scorer method (e.g. n_runs=10 for community, or sample_fraction=0.10 for edge support).	{}

Returns:

Type	Description
dict[tuple[int, int], float]	Canonical edge tuples (min(u, v), max(u, v)) → score.

Raises:

Type	Description
ValueError	If method is not recognised.

Examples:

>>> from spatial_graph_algorithms.simulate import generate
>>> from spatial_graph_algorithms.denoise import score_edges
>>> sg = generate(n=100, seed=0)
>>> scores = score_edges(sg, method="jaccard")
>>> len(scores) == sg.n_edges
True

Source code in src/spatial_graph_algorithms/denoise/__init__.py

def score_edges(
    sg: SpatialGraph,
    *,
    method: str,
    **kwargs: object,
) -> dict[tuple[int, int], float]:
    """Score all edges in *sg* using the specified method.

    Convenience wrapper around :class:`EdgeScorer`.

    Parameters
    ----------
    sg : SpatialGraph
        Input graph.
    method : str
        One of ``"betweenness"``, ``"jaccard"``, ``"square_bipartite"``,
        ``"community"``, ``"walk_visit_expansion"``, or
        ``"local_walk_support"``.
    **kwargs
        Forwarded to the scorer method (e.g. ``n_runs=10`` for community,
        or ``sample_fraction=0.10`` for edge support).

    Returns
    -------
    dict[tuple[int, int], float]
        Canonical edge tuples ``(min(u, v), max(u, v))`` → score.

    Raises
    ------
    ValueError
        If *method* is not recognised.

    Examples
    --------
    >>> from spatial_graph_algorithms.simulate import generate
    >>> from spatial_graph_algorithms.denoise import score_edges
    >>> sg = generate(n=100, seed=0)
    >>> scores = score_edges(sg, method="jaccard")
    >>> len(scores) == sg.n_edges
    True
    """
    method = _canonical_method(method)
    scorer = EdgeScorer()
    if not hasattr(scorer, method):
        raise ValueError(
            f"Unknown scoring method '{method}'. "
            f"Available: {_format_available_methods()}."
        )
    return getattr(scorer, method)(sg.adjacency_matrix, **kwargs)

spatial_graph_algorithms.denoise.denoise(sg, *, method, fraction_to_remove=0.05, **kwargs)

Score edges and remove the worst-scored fraction in one call.

Convenience wrapper combining :class:EdgeScorer with :meth:EdgeFilterer.by_fraction. Score polarity is looked up automatically from :data:SCORE_POLARITY.

Parameters:

Name	Type	Description	Default
sg	SpatialGraph	Input graph.	required
method	str	One of "betweenness", "jaccard", "square_bipartite", "community", "walk_visit_expansion", or "local_walk_support".	required
fraction_to_remove	float	Fraction of edges to remove, in [0, 1]. Default 0.05.	0.05
**kwargs	object	Forwarded to the scorer method.	{}

Returns:

Type	Description
SpatialGraph	New graph with the worst-scored edges removed and is_removed metadata set.

Examples:

>>> from spatial_graph_algorithms.simulate import generate
>>> from spatial_graph_algorithms.denoise import denoise
>>> sg = generate(n=200, false_edge_fraction=0.10, seed=42)
>>> sg_clean = denoise(sg, method="jaccard", fraction_to_remove=0.05)
>>> sg_clean.n_edges < sg.n_edges
True

Source code in src/spatial_graph_algorithms/denoise/__init__.py

def denoise(
    sg: SpatialGraph,
    *,
    method: str,
    fraction_to_remove: float = 0.05,
    **kwargs: object,
) -> SpatialGraph:
    """Score edges and remove the worst-scored fraction in one call.

    Convenience wrapper combining :class:`EdgeScorer` with
    :meth:`EdgeFilterer.by_fraction`.  Score polarity is looked up
    automatically from :data:`SCORE_POLARITY`.

    Parameters
    ----------
    sg : SpatialGraph
        Input graph.
    method : str
        One of ``"betweenness"``, ``"jaccard"``, ``"square_bipartite"``,
        ``"community"``, ``"walk_visit_expansion"``, or
        ``"local_walk_support"``.
    fraction_to_remove : float
        Fraction of edges to remove, in ``[0, 1]``.  Default 0.05.
    **kwargs
        Forwarded to the scorer method.

    Returns
    -------
    SpatialGraph
        New graph with the worst-scored edges removed and ``is_removed``
        metadata set.

    Examples
    --------
    >>> from spatial_graph_algorithms.simulate import generate
    >>> from spatial_graph_algorithms.denoise import denoise
    >>> sg = generate(n=200, false_edge_fraction=0.10, seed=42)
    >>> sg_clean = denoise(sg, method="jaccard", fraction_to_remove=0.05)
    >>> sg_clean.n_edges < sg.n_edges
    True
    """
    method = _canonical_method(method)
    scores = score_edges(sg, method=method, **kwargs)
    return EdgeFilterer().by_fraction(
        sg, scores, fraction=fraction_to_remove, method=method
    )

spatial_graph_algorithms.denoise.ES

Classes

EdgeSupportResult dataclass

Source code in src/spatial_graph_algorithms/denoise/ES.py

@dataclass
class EdgeSupportResult:
    edges: list[tuple[Any, Any]]
    scores: np.ndarray
    scored_mask: np.ndarray
    runtime_s: float
    config: EdgeSupportConfig

    def threshold_summary(self, threshold: float = 1.0, only_scored: bool = True) -> dict[str, Any]:
        """
        Summarise the threshold rule.

        For threshold=1:
            score < 1 means weaker support than expected.
            score > 1 means stronger support than expected.

        The estimated removal level is usually frac_below_threshold_scored.
        """
        scores = np.asarray(self.scores, dtype=np.float64)
        scored = np.asarray(self.scored_mask, dtype=bool)

        if only_scored:
            pool = scores[scored]
        else:
            pool = scores.copy()

        pool = pool[np.isfinite(pool)]

        if pool.size == 0:
            return {
                "threshold": float(threshold),
                "n_edges": int(len(scores)),
                "n_scored": int(scored.sum()),
                "n_pool": 0,
                "n_below_threshold": 0,
                "n_above_or_equal_threshold": 0,
                "frac_below_threshold": np.nan,
                "frac_above_or_equal_threshold": np.nan,
                "estimated_prune_fraction": np.nan,
            }

        below = pool < threshold
        above_or_equal = pool >= threshold

        n_below = int(below.sum())
        n_above_or_equal = int(above_or_equal.sum())

        return {
            "threshold": float(threshold),
            "n_edges": int(len(scores)),
            "n_scored": int(scored.sum()),
            "n_pool": int(pool.size),
            "n_below_threshold": n_below,
            "n_above_or_equal_threshold": n_above_or_equal,
            "frac_below_threshold": float(n_below / pool.size),
            "frac_above_or_equal_threshold": float(n_above_or_equal / pool.size),
            "estimated_prune_fraction": float(n_below / pool.size),
        }

    def suspicious_edges(
        self,
        threshold: float = 1.0,
        max_edges: Optional[int] = None,
    ) -> list[tuple[Any, Any, float]]:
        """
        Return scored edges with score < threshold, sorted from most suspicious upward.
        """
        rows = []
        for (u, v), s, ok in zip(self.edges, self.scores, self.scored_mask):
            if ok and np.isfinite(s) and s < threshold:
                rows.append((u, v, float(s)))

        rows.sort(key=lambda x: x[2])

        if max_edges is not None:
            rows = rows[: int(max_edges)]

        return rows

    def ranked_edges(self, only_scored: bool = True) -> list[tuple[Any, Any, float]]:
        """
        Return all edges sorted by ascending score.
        Lower score = more suspicious.
        """
        rows = []
        for (u, v), s, ok in zip(self.edges, self.scores, self.scored_mask):
            if only_scored and not ok:
                continue
            rows.append((u, v, float(s) if np.isfinite(s) else np.nan))

        rows.sort(key=lambda x: x[2] if np.isfinite(x[2]) else np.inf)
        return rows

    def to_dataframe(self) -> pd.DataFrame:
        """
        Convert result to a pandas DataFrame.
        Pandas is imported lazily so the core package does not require it unless used.
        """
        import pandas as pd

        return pd.DataFrame(
            {
                "u": [u for u, _ in self.edges],
                "v": [v for _, v in self.edges],
                "score": self.scores,
                "scored": self.scored_mask,
            }
        )

Methods:

threshold_summary(threshold=1.0, only_scored=True)

Summarise the threshold rule.

For threshold=1: score < 1 means weaker support than expected. score > 1 means stronger support than expected.

The estimated removal level is usually frac_below_threshold_scored.

Source code in src/spatial_graph_algorithms/denoise/ES.py

def threshold_summary(self, threshold: float = 1.0, only_scored: bool = True) -> dict[str, Any]:
    """
    Summarise the threshold rule.

    For threshold=1:
        score < 1 means weaker support than expected.
        score > 1 means stronger support than expected.

    The estimated removal level is usually frac_below_threshold_scored.
    """
    scores = np.asarray(self.scores, dtype=np.float64)
    scored = np.asarray(self.scored_mask, dtype=bool)

    if only_scored:
        pool = scores[scored]
    else:
        pool = scores.copy()

    pool = pool[np.isfinite(pool)]

    if pool.size == 0:
        return {
            "threshold": float(threshold),
            "n_edges": int(len(scores)),
            "n_scored": int(scored.sum()),
            "n_pool": 0,
            "n_below_threshold": 0,
            "n_above_or_equal_threshold": 0,
            "frac_below_threshold": np.nan,
            "frac_above_or_equal_threshold": np.nan,
            "estimated_prune_fraction": np.nan,
        }

    below = pool < threshold
    above_or_equal = pool >= threshold

    n_below = int(below.sum())
    n_above_or_equal = int(above_or_equal.sum())

    return {
        "threshold": float(threshold),
        "n_edges": int(len(scores)),
        "n_scored": int(scored.sum()),
        "n_pool": int(pool.size),
        "n_below_threshold": n_below,
        "n_above_or_equal_threshold": n_above_or_equal,
        "frac_below_threshold": float(n_below / pool.size),
        "frac_above_or_equal_threshold": float(n_above_or_equal / pool.size),
        "estimated_prune_fraction": float(n_below / pool.size),
    }

suspicious_edges(threshold=1.0, max_edges=None)

Return scored edges with score < threshold, sorted from most suspicious upward.

Source code in src/spatial_graph_algorithms/denoise/ES.py

def suspicious_edges(
    self,
    threshold: float = 1.0,
    max_edges: Optional[int] = None,
) -> list[tuple[Any, Any, float]]:
    """
    Return scored edges with score < threshold, sorted from most suspicious upward.
    """
    rows = []
    for (u, v), s, ok in zip(self.edges, self.scores, self.scored_mask):
        if ok and np.isfinite(s) and s < threshold:
            rows.append((u, v, float(s)))

    rows.sort(key=lambda x: x[2])

    if max_edges is not None:
        rows = rows[: int(max_edges)]

    return rows

ranked_edges(only_scored=True)

Return all edges sorted by ascending score. Lower score = more suspicious.

Source code in src/spatial_graph_algorithms/denoise/ES.py

def ranked_edges(self, only_scored: bool = True) -> list[tuple[Any, Any, float]]:
    """
    Return all edges sorted by ascending score.
    Lower score = more suspicious.
    """
    rows = []
    for (u, v), s, ok in zip(self.edges, self.scores, self.scored_mask):
        if only_scored and not ok:
            continue
        rows.append((u, v, float(s) if np.isfinite(s) else np.nan))

    rows.sort(key=lambda x: x[2] if np.isfinite(x[2]) else np.inf)
    return rows

to_dataframe()

Convert result to a pandas DataFrame. Pandas is imported lazily so the core package does not require it unless used.

Source code in src/spatial_graph_algorithms/denoise/ES.py

def to_dataframe(self) -> pd.DataFrame:
    """
    Convert result to a pandas DataFrame.
    Pandas is imported lazily so the core package does not require it unless used.
    """
    import pandas as pd

    return pd.DataFrame(
        {
            "u": [u for u, _ in self.edges],
            "v": [v for _, v in self.edges],
            "score": self.scores,
            "scored": self.scored_mask,
        }
    )

Functions:

local_walk_support(adj, *, walk_len=8, walks_per_edge=1000, gamma=0.85, min_targets=2, min_moves=2, require_both=False, agg='mean', sample_size=None, sample_fraction=None, seed=123, parallel=True, n_threads=None)

Score CSR adjacency edges by random-walk support. Low score = suspect.

Source code in src/spatial_graph_algorithms/denoise/ES.py

def local_walk_support(
    adj: scipy.sparse.csr_matrix,
    *,
    walk_len: int = 8,
    walks_per_edge: int = 1_000,
    gamma: float = 0.85,
    min_targets: int = 2,
    min_moves: int = 2,
    require_both: bool = False,
    agg: str = "mean",
    sample_size: int | None = None,
    sample_fraction: float | None = None,
    seed: int = 123,
    parallel: bool = True,
    n_threads: int | None = None,
) -> dict[tuple[int, int], float]:
    """Score CSR adjacency edges by random-walk support. Low score = suspect."""
    result = local_walk_support_result(
        adj,
        walk_len=walk_len,
        walks_per_edge=walks_per_edge,
        gamma=gamma,
        min_targets=min_targets,
        min_moves=min_moves,
        require_both=require_both,
        agg=agg,
        sample_size=sample_size,
        sample_fraction=sample_fraction,
        seed=seed,
        parallel=parallel,
        n_threads=n_threads,
    )
    scores: dict[tuple[int, int], float] = {}
    for (u, v), s, ok in zip(result.edges, result.scores, result.scored_mask):
        if not ok or not np.isfinite(s):
            continue
        cu = int(u)
        cv = int(v)
        if cu == cv:
            continue
        key = (cu, cv) if cu < cv else (cv, cu)
        scores[key] = float(s)
    return scores

edge_support(adj, *, walk_len=8, walks_per_edge=1000, gamma=0.85, min_targets=2, min_moves=2, require_both=False, agg='mean', sample_size=None, sample_fraction=None, seed=123, parallel=True, n_threads=None)

Compatibility alias for :func:local_walk_support.

Source code in src/spatial_graph_algorithms/denoise/ES.py

def edge_support(
    adj: scipy.sparse.csr_matrix,
    *,
    walk_len: int = 8,
    walks_per_edge: int = 1_000,
    gamma: float = 0.85,
    min_targets: int = 2,
    min_moves: int = 2,
    require_both: bool = False,
    agg: str = "mean",
    sample_size: int | None = None,
    sample_fraction: float | None = None,
    seed: int = 123,
    parallel: bool = True,
    n_threads: int | None = None,
) -> dict[tuple[int, int], float]:
    """Compatibility alias for :func:`local_walk_support`."""
    return local_walk_support(
        adj,
        walk_len=walk_len,
        walks_per_edge=walks_per_edge,
        gamma=gamma,
        min_targets=min_targets,
        min_moves=min_moves,
        require_both=require_both,
        agg=agg,
        sample_size=sample_size,
        sample_fraction=sample_fraction,
        seed=seed,
        parallel=parallel,
        n_threads=n_threads,
    )

local_walk_support_result(adj, *, walk_len=8, walks_per_edge=1000, gamma=0.85, min_targets=2, min_moves=2, require_both=False, agg='mean', sample_size=None, sample_fraction=None, seed=123, parallel=True, n_threads=None)

Return full local-walk-support diagnostics for a CSR adjacency matrix.

Source code in src/spatial_graph_algorithms/denoise/ES.py

def local_walk_support_result(
    adj: scipy.sparse.csr_matrix,
    *,
    walk_len: int = 8,
    walks_per_edge: int = 1_000,
    gamma: float = 0.85,
    min_targets: int = 2,
    min_moves: int = 2,
    require_both: bool = False,
    agg: str = "mean",
    sample_size: int | None = None,
    sample_fraction: float | None = None,
    seed: int = 123,
    parallel: bool = True,
    n_threads: int | None = None,
) -> EdgeSupportResult:
    """Return full local-walk-support diagnostics for a CSR adjacency matrix."""
    if not isinstance(adj, scipy.sparse.csr_matrix):
        raise TypeError("adj must be a csr_matrix")
    if adj.shape[0] != adj.shape[1]:
        raise ValueError("adj must be square")

    try:
        graph = nx.from_scipy_sparse_array(adj, create_using=nx.Graph)
    except Exception as exc:
        raise ValueError(f"Failed to convert adjacency to graph: {exc}") from exc

    config = EdgeSupportConfig(
        walk_len=walk_len,
        walks_per_edge=walks_per_edge,
        gamma=gamma,
        min_targets=min_targets,
        min_moves=min_moves,
        require_both=require_both,
        agg=agg,  # type: ignore[arg-type]
        weight="weight",
        parallel=parallel,
        n_threads=n_threads,
    )
    return score_local_walk_support(
        graph,
        config=config,
        edges_to_score=None,
        sample_size=sample_size,
        sample_fraction=sample_fraction,
        seed=seed,
    )

edge_support_result(adj, *, walk_len=8, walks_per_edge=1000, gamma=0.85, min_targets=2, min_moves=2, require_both=False, agg='mean', sample_size=None, sample_fraction=None, seed=123, parallel=True, n_threads=None)

Compatibility alias for :func:local_walk_support_result.

Source code in src/spatial_graph_algorithms/denoise/ES.py

def edge_support_result(
    adj: scipy.sparse.csr_matrix,
    *,
    walk_len: int = 8,
    walks_per_edge: int = 1_000,
    gamma: float = 0.85,
    min_targets: int = 2,
    min_moves: int = 2,
    require_both: bool = False,
    agg: str = "mean",
    sample_size: int | None = None,
    sample_fraction: float | None = None,
    seed: int = 123,
    parallel: bool = True,
    n_threads: int | None = None,
) -> EdgeSupportResult:
    """Compatibility alias for :func:`local_walk_support_result`."""
    return local_walk_support_result(
        adj,
        walk_len=walk_len,
        walks_per_edge=walks_per_edge,
        gamma=gamma,
        min_targets=min_targets,
        min_moves=min_moves,
        require_both=require_both,
        agg=agg,
        sample_size=sample_size,
        sample_fraction=sample_fraction,
        seed=seed,
        parallel=parallel,
        n_threads=n_threads,
    )

score_local_walk_support(graph_or_edges, config=None, edges_to_score=None, sample_size=None, sample_fraction=None, seed=123)

Score graph edges by local random-walk support.

Parameters:

Name	Type	Description	Default
graph_or_edges	Graph | Iterable[tuple[Any, Any]]	Either a NetworkX graph or an iterable of edges. If an edge list is supplied, an undirected NetworkX graph is built from it.	required
config	Optional[EdgeSupportConfig]	EdgeSupportConfig object.	None
edges_to_score	Optional[Iterable[tuple[Any, Any]]]	Optional subset of edges to score. If None, all graph edges are candidates.	None
sample_size	Optional[int]	Optional number of candidate edges to sample before scoring.	None
sample_fraction	Optional[float]	Optional fraction of candidate edges to sample before scoring.	None
seed	int	Sampling seed. The walk RNG seed is controlled by config through deterministic edge seeds.	123

Returns:

Type	Description
EdgeSupportResult

Source code in src/spatial_graph_algorithms/denoise/ES.py

def score_local_walk_support(
    graph_or_edges: nx.Graph | Iterable[tuple[Any, Any]],
    config: Optional[EdgeSupportConfig] = None,
    edges_to_score: Optional[Iterable[tuple[Any, Any]]] = None,
    sample_size: Optional[int] = None,
    sample_fraction: Optional[float] = None,
    seed: int = 123,
) -> EdgeSupportResult:
    """
    Score graph edges by local random-walk support.

    Parameters
    ----------
    graph_or_edges:
        Either a NetworkX graph or an iterable of edges.
        If an edge list is supplied, an undirected NetworkX graph is built from it.

    config:
        EdgeSupportConfig object.

    edges_to_score:
        Optional subset of edges to score.
        If None, all graph edges are candidates.

    sample_size:
        Optional number of candidate edges to sample before scoring.

    sample_fraction:
        Optional fraction of candidate edges to sample before scoring.

    seed:
        Sampling seed. The walk RNG seed is controlled by config through deterministic edge seeds.

    Returns
    -------
    EdgeSupportResult
    """
    if config is None:
        config = EdgeSupportConfig()

    _validate_config(config)

    if config.n_threads is not None:
        set_num_threads(int(config.n_threads))

    G = _as_graph(graph_or_edges, weight_attr=config.weight)

    if G.number_of_edges() == 0:
        raise ValueError("Cannot score an empty graph.")

    candidate_edges = _resolve_candidate_edges(G, edges_to_score)
    candidate_edges = _sample_edges(
        candidate_edges,
        sample_size=sample_size,
        sample_fraction=sample_fraction,
        seed=seed,
    )

    if len(candidate_edges) == 0:
        raise ValueError("No candidate edges to score.")

    t0 = time.perf_counter()

    node_order, node_to_idx, indices, indptr, weights, deg_w, total_deg_w = _graph_to_arrays(
        G,
        weight_attr=config.weight,
    )

    pair_idx = []
    kept_edges = []

    for u, v in candidate_edges:
        if u not in node_to_idx or v not in node_to_idx:
            continue
        if not G.has_edge(u, v):
            continue

        iu = node_to_idx[u]
        iv = node_to_idx[v]

        if iu == iv:
            continue

        pair_idx.append((iu, iv))
        kept_edges.append((u, v))

    if len(pair_idx) == 0:
        raise ValueError("None of the requested edges exist in the graph.")

    pair_idx = np.asarray(pair_idx, dtype=np.int32)

    directed_u = np.empty(2 * len(pair_idx), dtype=np.int32)
    directed_v = np.empty(2 * len(pair_idx), dtype=np.int32)

    directed_u[0::2] = pair_idx[:, 0]
    directed_v[0::2] = pair_idx[:, 1]

    directed_u[1::2] = pair_idx[:, 1]
    directed_v[1::2] = pair_idx[:, 0]

    if config.parallel:
        directed_scores = _score_directed_pairs_parallel(
            directed_u,
            directed_v,
            indices,
            indptr,
            weights,
            deg_w,
            float(total_deg_w),
            int(config.walk_len),
            int(config.walks_per_edge),
            float(config.gamma),
            np.uint64(seed),
            int(config.min_targets),
            int(config.min_moves),
        )
    else:
        directed_scores = _score_directed_pairs_serial(
            directed_u,
            directed_v,
            indices,
            indptr,
            weights,
            deg_w,
            float(total_deg_w),
            int(config.walk_len),
            int(config.walks_per_edge),
            float(config.gamma),
            np.uint64(seed),
            int(config.min_targets),
            int(config.min_moves),
        )

    scores = _aggregate_directed_scores(
        directed_scores,
        require_both=config.require_both,
        agg=config.agg,
    )

    scored_mask = np.isfinite(scores)
    runtime_s = time.perf_counter() - t0

    return EdgeSupportResult(
        edges=kept_edges,
        scores=scores.astype(np.float64),
        scored_mask=scored_mask,
        runtime_s=float(runtime_s),
        config=config,
    )

score_edge_support(graph_or_edges, config=None, edges_to_score=None, sample_size=None, sample_fraction=None, seed=123)

Compatibility alias for :func:score_local_walk_support.

Source code in src/spatial_graph_algorithms/denoise/ES.py

def score_edge_support(
    graph_or_edges: nx.Graph | Iterable[tuple[Any, Any]],
    config: Optional[EdgeSupportConfig] = None,
    edges_to_score: Optional[Iterable[tuple[Any, Any]]] = None,
    sample_size: Optional[int] = None,
    sample_fraction: Optional[float] = None,
    seed: int = 123,
) -> EdgeSupportResult:
    """Compatibility alias for :func:`score_local_walk_support`."""
    return score_local_walk_support(
        graph_or_edges,
        config=config,
        edges_to_score=edges_to_score,
        sample_size=sample_size,
        sample_fraction=sample_fraction,
        seed=seed,
    )

estimate_threshold_prune_fraction(result, threshold=1.0)

Convenience wrapper.

Returns the fraction of scored edges with score < threshold. For threshold=1, this is the local-walk-support estimate of suspicious-edge mass.

Source code in src/spatial_graph_algorithms/denoise/ES.py

def estimate_threshold_prune_fraction(
    result: EdgeSupportResult,
    threshold: float = 1.0,
) -> float:
    """
    Convenience wrapper.

    Returns the fraction of scored edges with score < threshold.
    For threshold=1, this is the local-walk-support estimate of suspicious-edge mass.
    """
    summary = result.threshold_summary(threshold=threshold, only_scored=True)
    return float(summary["estimated_prune_fraction"])

prune_edges_below_threshold(G, result, threshold=1.0, min_degree=3, copy=True)

Remove scored edges with score < threshold, using a degree guard.

Lower score = more suspicious.

Source code in src/spatial_graph_algorithms/denoise/ES.py

def prune_edges_below_threshold(
    G: nx.Graph,
    result: EdgeSupportResult,
    threshold: float = 1.0,
    min_degree: int = 3,
    copy: bool = True,
) -> tuple[nx.Graph, dict[str, Any]]:
    """
    Remove scored edges with score < threshold, using a degree guard.

    Lower score = more suspicious.
    """
    G2 = G.copy() if copy else G

    ranked = result.suspicious_edges(threshold=threshold)
    requested = len(ranked)

    pruned = 0
    skipped_missing = 0
    skipped_degree_guard = 0

    removed_edges = []

    for u, v, score in ranked:
        if not G2.has_edge(u, v):
            skipped_missing += 1
            continue

        if G2.degree(u) > min_degree and G2.degree(v) > min_degree:
            G2.remove_edge(u, v)
            pruned += 1
            removed_edges.append((u, v, score))
        else:
            skipped_degree_guard += 1

    info = {
        "threshold": float(threshold),
        "requested": int(requested),
        "pruned": int(pruned),
        "skipped_missing": int(skipped_missing),
        "skipped_degree_guard": int(skipped_degree_guard),
        "min_degree": int(min_degree),
        "hit_rate": float(pruned / max(1, requested)),
        "removed_edges": removed_edges,
    }

    return G2, info

downweight_edges_below_threshold(G, result, threshold=1.0, min_mult=0.05, weight='weight', copy=True)

Linearly downweight edges with score < threshold.

The most suspicious edge receives multiplier min_mult. The least suspicious edge below the threshold receives multiplier near 1.

Source code in src/spatial_graph_algorithms/denoise/ES.py

def downweight_edges_below_threshold(
    G: nx.Graph,
    result: EdgeSupportResult,
    threshold: float = 1.0,
    min_mult: float = 0.05,
    weight: str = "weight",
    copy: bool = True,
) -> tuple[nx.Graph, dict[str, Any]]:
    """
    Linearly downweight edges with score < threshold.

    The most suspicious edge receives multiplier min_mult.
    The least suspicious edge below the threshold receives multiplier near 1.
    """
    G2 = G.copy() if copy else G

    ranked = result.suspicious_edges(threshold=threshold)
    n = len(ranked)

    if n == 0:
        return G2, {
            "threshold": float(threshold),
            "requested": 0,
            "downweighted": 0,
            "min_mult": float(min_mult),
        }

    downweighted = 0
    skipped_missing = 0

    denom = max(1, n - 1)

    for rank, (u, v, score) in enumerate(ranked):
        if not G2.has_edge(u, v):
            skipped_missing += 1
            continue

        frac = rank / denom
        mult = float(min_mult + (1.0 - min_mult) * frac)

        d = G2.edges[u, v]
        w0 = float(d.get(weight, 1.0))

        if f"{weight}_orig" not in d:
            d[f"{weight}_orig"] = w0

        d[weight] = w0 * mult
        d["local_walk_support_score"] = float(score)
        d["local_walk_support_downweight_mult"] = float(mult)
        d["edge_support_score"] = float(score)
        d["edge_support_downweight_mult"] = float(mult)

        downweighted += 1

    info = {
        "threshold": float(threshold),
        "requested": int(n),
        "downweighted": int(downweighted),
        "skipped_missing": int(skipped_missing),
        "min_mult": float(min_mult),
    }

    return G2, info