Algorithm Search⌘K

E2LSH in Python

LSH for Euclidean distance built from p-stable random projections quantized into buckets.

File
e2lsh.py, 68 lines
Level
I0 not checked yet
Toolchain
Python 3.14, pytest 9
Test framework
pytest
Run the tests
python3 -m pytest
e2lsh.pyRawGitHub
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/e2lsh/. The authors are listed in prov/K84C3G.jsonl.

import math
import random
from collections.abc import Sequence

Vector = Sequence[float]


def euclidean(x: Vector, y: Vector) -> float:
    return math.sqrt(sum((a - b) * (a - b) for a, b in zip(x, y)))


def collision_probability(u: float, w: float) -> float:
    """Pr[h(x) = h(y)] for ||x - y|| = u under h(v) = floor((a.v + b)/w), a Gaussian, b uniform in [0, w)."""
    if u == 0:
        return 1.0
    t = w / u
    phi = 0.5 * (1 + math.erf(-t / math.sqrt(2)))
    return 1 - 2 * phi - 2 / (math.sqrt(2 * math.pi) * t) * (1 - math.exp(-t * t / 2))


def parameters(n: int, w: float, r: float, c: float) -> tuple[int, int]:
    """k and L of the LSH theorem for the (r, cr, p(r), p(cr)) family."""
    p1, p2 = collision_probability(r, w), collision_probability(c * r, w)
    rho = math.log(1 / p1) / math.log(1 / p2)
    k = max(1, math.ceil(math.log(n) / math.log(1 / p2)))
    return k, math.ceil(n**rho / p1)


class E2LSH:
    """LSH for the Euclidean distance with quantized Gaussian projections.

    Build: O(n L k d). Query: O(L k d) for the keys plus at most 3L distances of O(d) each.
    """

    def __init__(
        self, points: Sequence[Vector], k: int, L: int, w: float, rng: random.Random | None = None
    ) -> None:
        rng = rng if rng is not None else random.Random()
        self.points = [tuple(p) for p in points]
        d = len(self.points[0]) if self.points else 0
        self.L, self.w = L, w
        self.funcs: list[list[tuple[list[float], float]]] = []
        self.tables: list[dict[tuple[int, ...], list[int]]] = []
        for _ in range(L):  # @line L1
            g = [([rng.gauss(0, 1) for _ in range(d)], rng.uniform(0, w)) for _ in range(k)]  # @line L2
            self.funcs.append(g)
            table: dict[tuple[int, ...], list[int]] = {}
            for i, x in enumerate(self.points):
                table.setdefault(self.key(g, x), []).append(i)  # @line L3
            self.tables.append(table)

    def key(self, g: list[tuple[list[float], float]], v: Vector) -> tuple[int, ...]:
        return tuple(math.floor((sum(ai * vi for ai, vi in zip(a, v)) + b) / self.w) for a, b in g)

    def query(self, q: Vector, c: float, r: float) -> int | None:
        """Return the index of a point within c*r of q, or None."""
        count = 0  # @line L4
        for j in range(self.L):  # @line L5
            for i in self.tables[j].get(self.key(self.funcs[j], q), []):
                count += 1  # @line L6
                if euclidean(q, self.points[i]) <= c * r:
                    return i
                if count >= 3 * self.L:  # @line L7
                    return None
        return None  # @line L8
test_e2lsh.py 65 lines
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/e2lsh/. The authors are listed in prov/K84C3G.jsonl.

import math
import random

from e2lsh import E2LSH, collision_probability, euclidean, parameters


def test_known_values() -> None:
    assert abs(collision_probability(1, 4) - 0.8005) < 1e-4
    assert abs(collision_probability(2, 4) - 0.6095) < 1e-4
    assert collision_probability(0, 4) == 1.0
    # p depends only on w/u, and decreases in u
    assert abs(collision_probability(2, 8) - collision_probability(1, 4)) < 1e-12
    ps = [collision_probability(u / 10, 4) for u in range(1, 200)]
    assert all(a > b for a, b in zip(ps, ps[1:]))


def test_rho_below_one_over_c_for_w4() -> None:
    for c in (1.5, 2, 3, 4):
        p1, p2 = collision_probability(1, 4), collision_probability(c, 4)
        assert math.log(1 / p1) / math.log(1 / p2) < 1 / c


def test_collision_probability_monte_carlo() -> None:
    rng = random.Random(11)
    w = 4.0
    for u in (0.5, 1.0, 2.0, 5.0):
        hits = 0
        trials = 20000
        for _ in range(trials):
            a, b = rng.gauss(0, 1), rng.uniform(0, w)
            hits += math.floor(b / w) == math.floor((a * u + b) / w)
        assert abs(hits / trials - collision_probability(u, w)) < 0.015


def test_answers_valid_against_brute_force() -> None:
    rng = random.Random(123)
    for _ in range(500):
        d = rng.randint(1, 4)
        pts = [[rng.uniform(-5, 5) for _ in range(d)] for _ in range(rng.randint(0, 20))]
        idx = E2LSH(pts, rng.randint(1, 3), rng.randint(1, 3), rng.choice((1.0, 2.0, 4.0)), rng)
        q = [rng.uniform(-5, 5) for _ in range(d)]
        r, c = rng.uniform(0.1, 2), 2.0
        got = idx.query(q, c, r)
        if got is not None:
            assert euclidean(q, pts[got]) <= c * r
        if not pts or min(euclidean(q, p) for p in pts) > c * r:
            assert got is None


def test_success_rate() -> None:
    rng = random.Random(6)
    n, d, r, c, w = 50, 8, 1.0, 2.0, 4.0
    k, L = parameters(n, w, r, c)
    ok = 0
    trials = 200
    for _ in range(trials):
        pts = [[rng.gauss(0, 3) for _ in range(d)] for _ in range(n)]
        direction = [rng.gauss(0, 1) for _ in range(d)]
        norm = math.sqrt(sum(x * x for x in direction))
        q = [p + 0.99 * r * x / norm for p, x in zip(pts[0], direction)]
        ok += E2LSH(pts, k, L, w, rng).query(q, c, r) is not None
    assert ok / trials >= 2 / 3 - 1 / math.e

What the next level needs

  • the tests did not run in this build

Read the full page about E2LSH