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
# 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